diff --git a/src/sentry/workflow_engine/models/workflow.py b/src/sentry/workflow_engine/models/workflow.py
index 1264813ea1d6..90bb17071657 100644
--- a/src/sentry/workflow_engine/models/workflow.py
+++ b/src/sentry/workflow_engine/models/workflow.py
@@ -125,7 +125,10 @@ def get_snapshot(self) -> WorkflowSnapshot:
         }
 
     def evaluate_trigger_conditions(
-        self, event_data: WorkflowEventData, when_data_conditions: list[DataCondition] | None = None
+        self,
+        event_data: WorkflowEventData,
+        when_data_conditions: list[DataCondition] | None = None,
+        group: DataConditionGroup | None = None,
     ) -> tuple[TriggerResult, list[DataCondition]]:
         """
         Evaluate the conditions for the workflow trigger and return if the evaluation was successful.
@@ -140,12 +143,11 @@ def evaluate_trigger_conditions(
             return TriggerResult.TRUE, []
 
         workflow_event_data = replace(event_data, workflow_env=self.environment)
-        try:
-            group = DataConditionGroup.objects.get_from_cache(id=self.when_condition_group_id)
-        except DataConditionGroup.DoesNotExist:
+
+        if group is None:
             # This isn't expected under normal conditions, but weird things can happen in the
             # midst of deletions and migrations.
-            logger.exception(
+            logger.critical(
                 "DataConditionGroup does not exist",
                 extra={"id": self.when_condition_group_id},
             )
diff --git a/src/sentry/workflow_engine/processors/workflow.py b/src/sentry/workflow_engine/processors/workflow.py
index c0bd22cd3334..b1ae831974fd 100644
--- a/src/sentry/workflow_engine/processors/workflow.py
+++ b/src/sentry/workflow_engine/processors/workflow.py
@@ -156,14 +156,21 @@ def evaluate_workflow_triggers(
     # Retrieve these as a batch to avoid a query/cache-lookup per DCG.
     data_conditions_by_dcg_id = _get_data_conditions_for_group_by_dcg(dcg_ids)
 
+    # Retrieve data condition groups as a batch to avoid a query/cache-lookup per DCG.
+    data_condition_groups_by_id: dict[int, DataConditionGroup] = {
+        dcg.id: dcg for dcg in DataConditionGroup.objects.get_many_from_cache(dcg_ids)
+    }
+
     tainted_untriggered, untainted_untriggered = 0, 0
     for workflow in workflows:
         when_data_conditions = None
+        when_condition_group = None
         if dcg_id := workflow.when_condition_group_id:
             when_data_conditions = data_conditions_by_dcg_id.get(dcg_id)
+            when_condition_group = data_condition_groups_by_id.get(dcg_id)
 
         evaluation, remaining_conditions = workflow.evaluate_trigger_conditions(
-            event_data, when_data_conditions
+            event_data, when_data_conditions, when_condition_group
         )
 
         if remaining_conditions:
diff --git a/tests/sentry/workflow_engine/models/test_workflow.py b/tests/sentry/workflow_engine/models/test_workflow.py
index 0df0e8d685ee..fe209b5271d8 100644
--- a/tests/sentry/workflow_engine/models/test_workflow.py
+++ b/tests/sentry/workflow_engine/models/test_workflow.py
@@ -31,21 +31,27 @@ def test_queryset(self) -> None:
         assert not Workflow.objects.filter(id=self.workflow.id).exists()
 
     def test_evaluate_trigger_conditions__condition_new_event__True(self) -> None:
-        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.event_data)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(
+            self.event_data, group=self.data_condition_group
+        )
         assert evaluation.triggered is True
 
     def test_evaluate_trigger_conditions__condition_new_event__False(self) -> None:
         # Update event to have been seen before
         self.group_event.group.times_seen = 5
 
-        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.event_data)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(
+            self.event_data, group=self.data_condition_group
+        )
         assert evaluation.triggered is False
 
     def test_evaluate_trigger_conditions__no_conditions(self) -> None:
         self.workflow.when_condition_group = None
         self.workflow.save()
 
-        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.event_data)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(
+            self.event_data, group=self.data_condition_group
+        )
         assert evaluation.triggered is True
 
     def test_evaluate_trigger_conditions__slow_condition(self) -> None:
@@ -57,7 +63,7 @@ def test_evaluate_trigger_conditions__slow_condition(self) -> None:
         )
         self.data_condition_group.conditions.add(slow_condition)
         evaluation, remaining_conditions = self.workflow.evaluate_trigger_conditions(
-            self.event_data
+            self.event_data, group=self.data_condition_group
         )
 
         assert evaluation.triggered is True