AAP-87084 — Replace UNION with OR + pre-computed role set in unified job RBAC query (#16577)

* AAP-87084 — Replace UNION with OR + pre-computed role set in unified job RBAC query

The UNION ALL approach introduced by AAP-81173 forces PostgreSQL to
materialize all accessible job IDs across four RBAC branches before
any filtering or LIMIT can apply. This causes a 10x per-call regression
on the unified jobs list (36ms → 366ms) and a 3x regression on the
dashboard date-bucketed aggregation (88.9ms → 261.5ms).

Pre-compute the user's role IDs once as a Python list and pass them as
literal parameters to OR-based RoleEvaluation filters. This eliminates
the 4x redundant roleuserassignment subquery scans and restores
single-pass filtering with early LIMIT exit.

Resolves: AAP-87084, AAP-87087

* Assert RBAC query list is non-empty before checking for UNION

Ensures the test does not pass vacuously if no RBAC query is captured.

* Add test for singleton permission shortcut branches

Exercises the UJT, inventory, and org auditor singleton permission
paths in filtered_queryset() to increase coverage on new code.

* Add test for team-granted permissions in unified job list

Verifies that a user who can view a job template only through a team
assignment (not direct) sees the corresponding unified jobs.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

awx/main/access.py

@@ -12,6 +12,7 @@ from django.conf import settings
 from django.db.models import Q, Prefetch
 from django.contrib.auth.models import User
 from django.utils.translation import gettext_lazy as _
+from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist
 
 # Django REST Framework
@@ -19,7 +20,7 @@ from rest_framework.exceptions import ParseError, PermissionDenied
 
 # django-ansible-base
 from ansible_base.lib.utils.validation import to_python_boolean
-from ansible_base.rbac.models import RoleEvaluation
+from ansible_base.rbac.models import RoleEvaluation, RoleUserAssignment
 from ansible_base.rbac.policies import visible_users
 from ansible_base.rbac import permission_registry
 
@@ -2507,39 +2508,75 @@ class UnifiedJobAccess(BaseAccess):
     # )
 
     def filtered_queryset(self):
-        inv_pk_qs = Inventory.access_ids_qs(self.user, 'view')
+        # AAP-87084 / AAP-87087: Pre-compute role IDs once to avoid 4x redundant
+        # roleuserassignment scans, and use OR (not UNION) to allow single-pass
+        # filtering with early LIMIT exit.
+        user_role_ids = list(RoleUserAssignment.objects.filter(user_id=self.user.id).values_list('object_role_id', flat=True))
+        if not user_role_ids:
+            return self.model.objects.none()
 
-        by_template = (
+        user_singletons = self.user.singleton_permissions()
-            self.model.objects.filter(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
+
-            .order_by()
+        role_subclasses = UnifiedJobTemplate._submodels_with_roles()
-            .values_list('pk', flat=True)
+        ujt_codenames = [f'view_{cls._meta.model_name}' for cls in role_subclasses]
+        if not (set(ujt_codenames) - user_singletons):
+            ujt_accessible = UnifiedJobTemplate.objects.filter(polymorphic_ctype__in=ContentType.objects.get_for_models(*role_subclasses).values()).values_list(
+                'id', flat=True
+            )
+        else:
+            dab_role_cts = permission_registry.content_type_model.objects.get_for_models(*role_subclasses).values()
+            ujt_accessible = (
+                RoleEvaluation.objects.filter(
+                    role_id__in=user_role_ids,
+                    codename__in=ujt_codenames,
+                    content_type_id__in=[ct.id for ct in dab_role_cts],
+                )
+                .values_list('object_id')
+                .distinct()
             )
 
-        by_inventory_update = (
+        if 'view_inventory' in user_singletons:
-            InventoryUpdate.objects.filter(
+            inv_accessible = Inventory.objects.values_list('id', flat=True)
-                inventory_source__inventory__id__in=inv_pk_qs,
+        else:
+            inv_ct_id = permission_registry.content_type_model.objects.get_for_model(Inventory).id
+            inv_accessible = (
+                RoleEvaluation.objects.filter(
+                    role_id__in=user_role_ids,
+                    codename='view_inventory',
+                    content_type_id=inv_ct_id,
                 )
-            .order_by()
+                .values_list('object_id')
-            .values_list('pk', flat=True)
+                .distinct()
             )
 
-        by_adhoc = (
+        if 'audit_organization' in user_singletons:
-            AdHocCommand.objects.filter(
+            org_accessible = Organization.objects.values_list('id', flat=True)
-                inventory__id__in=inv_pk_qs,
+        else:
+            org_ct_id = permission_registry.content_type_model.objects.get_for_model(Organization).id
+            org_accessible = (
+                RoleEvaluation.objects.filter(
+                    role_id__in=user_role_ids,
+                    codename='audit_organization',
+                    content_type_id=org_ct_id,
                 )
-            .order_by()
+                .values_list('object_id')
-            .values_list('pk', flat=True)
+                .distinct()
             )
 
-        by_org_auditor = (
+        return self.model.objects.filter(
-            self.model.objects.filter(
+            Q(unified_job_template_id__in=ujt_accessible)
-                organization__in=Organization.access_ids_qs(self.user, 'audit_organization'),
+            | Q(
+                pk__in=InventoryUpdate.objects.filter(
+                    inventory_source__inventory__id__in=inv_accessible,
+                ).values('pk')
             )
-            .order_by()
+            | Q(
-            .values_list('pk', flat=True)
+                pk__in=AdHocCommand.objects.filter(
+                    inventory__id__in=inv_accessible,
+                ).values('pk')
+            )
+            | Q(organization__in=org_accessible)
         )
-
-        return self.model.objects.filter(pk__in=by_template.union(by_inventory_update, by_adhoc, by_org_auditor))
 
     def get_queryset(self):
         return super(UnifiedJobAccess, self).get_queryset().filter(workflowapproval__isnull=True)

awx/main/tests/functional/dab_rbac/test_unified_job_access.py

@@ -13,13 +13,14 @@ from awx.main.models import (
     JobTemplate,
     Organization,
     Project,
+    Team,
     UnifiedJob,
 )
 
 
 @pytest.mark.django_db
-def test_unified_job_list_uses_union(user, organization, inventory, setup_managed_roles, get):
+def test_unified_job_list_uses_or_not_union(user, organization, inventory, setup_managed_roles, get):
-    """The unified job list RBAC query uses UNION instead of OR to allow per-branch query planning."""
+    """The unified job list RBAC query uses OR-based filtering, not UNION."""
     org_admin = user('uj-org-admin')
     RoleDefinition.objects.get(name='Organization Admin').give_permission(org_admin, organization)
 
@@ -38,8 +39,10 @@ def test_unified_job_list_uses_union(user, organization, inventory, setup_manage
     assert response.status_code == 200
     assert response.data['count'] >= 3
 
-    uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'UNION' in q['sql'] and 'main_unifiedjob' in q['sql']]
+    uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'main_unifiedjob' in q['sql'] and 'dab_rbac_roleevaluation' in q['sql']]
-    assert uj_rbac_queries, "Expected at least one query using UNION for unified job RBAC filtering"
+    assert uj_rbac_queries, "Expected a unified-job RBAC query"
+    for sql in uj_rbac_queries:
+        assert 'UNION' not in sql, "RBAC query should use OR, not UNION"
 
 
 @pytest.mark.django_db
@@ -77,6 +80,62 @@ def test_unified_job_list_inventory_viewer_sees_inventory_updates(user, setup_ma
     assert inv_update.pk in result_ids
 
 
+@pytest.mark.django_db
+def test_unified_job_list_singleton_permissions(user, organization, inventory, setup_managed_roles, get):
+    """Users with global (singleton) view permissions see jobs via shortcut paths
+    that bypass RoleEvaluation queries entirely."""
+    singleton_user = user('uj-singleton')
+    RoleDefinition.objects.get(name='Organization Admin').give_permission(singleton_user, organization)
+
+    project = Project.objects.create(name='uj-singleton-project', organization=organization)
+    jt = JobTemplate.objects.create(name='uj-singleton-jt', project=project, inventory=inventory, organization=organization)
+    job = jt.create_unified_job()
+
+    inv_src = InventorySource.objects.create(name='uj-singleton-invsrc', inventory=inventory, source='ec2')
+    inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source)
+
+    adhoc = AdHocCommand.objects.create(name='uj-singleton-adhoc', inventory=inventory)
+
+    # Inject singleton permissions to exercise the shortcut branches in
+    # filtered_queryset() without needing a global RoleDefinition.
+    singleton_user._singleton_permissions = {
+        'view_jobtemplate',
+        'view_project',
+        'view_workflowjobtemplate',
+        'view_inventory',
+        'audit_organization',
+    }
+
+    response = get(reverse('api:unified_job_list'), singleton_user)
+    assert response.status_code == 200
+    result_ids = [r['id'] for r in response.data['results']]
+    assert job.pk in result_ids
+    assert inv_update.pk in result_ids
+    assert adhoc.pk in result_ids
+
+
+@pytest.mark.django_db
+def test_unified_job_list_team_member_sees_team_granted_jobs(user, setup_managed_roles, get):
+    """A user who can view a JT only through a team assignment must see
+    the corresponding unified jobs in the list."""
+    org = Organization.objects.create(name='uj-team-org')
+    team = Team.objects.create(name='uj-test-team', organization=org)
+    team_user = user('uj-team-member')
+
+    RoleDefinition.objects.get(name='Team Member').give_permission(team_user, team)
+
+    inventory = org.inventories.create(name='uj-team-inv')
+    project = Project.objects.create(name='uj-team-project', organization=org)
+    jt = JobTemplate.objects.create(name='uj-team-jt', project=project, inventory=inventory, organization=org)
+    RoleDefinition.objects.get(name='JobTemplate Execute').give_permission(team, jt)
+    job = jt.create_unified_job()
+
+    response = get(reverse('api:unified_job_list'), team_user)
+    assert response.status_code == 200
+    result_ids = [r['id'] for r in response.data['results']]
+    assert job.pk in result_ids, f"Team member should see job {job.pk} via team-granted JT execute permission, " f"but got result IDs: {result_ids}"
+
+
 @pytest.mark.django_db
 def test_unified_job_list_rando_sees_nothing(rando, setup_managed_roles, get):
     """Unprivileged user sees no unified jobs."""