Keep backfill counts consistent with corrupt row skips

Author: reachsrinivaschennupati-tech

python_modules/dagster/dagster/_core/storage/runs/sql_run_storage.py

@@ -1087,22 +1087,17 @@ def _filter_backfill_candidates(
 
     def get_backfills_count(self, filters: BulkActionsFilter | None = None) -> int:
         check.opt_inst_param(filters, "filters", BulkActionsFilter)
+        query = self._backfills_query(filters=filters)
+        rows = self.fetchall(query)
+        backfill_candidates = self._deserialize_backfill_rows(rows)
         if filters and filters.tags:
             # runs can have more tags than the backfill that launched them. Since we filtered tags by
             # querying for runs with those tags, we need to do an additional check that the backfills
             # also have the requested tags. This requires fetching the backfills from the db and filtering them
-            query = self._backfills_query(filters=filters)
-            rows = self.fetchall(query)
-            backfill_candidates = self._deserialize_backfill_rows(rows)
-            return len(
-                self._apply_backfill_tags_filter_to_results(backfill_candidates, filters.tags)
+            backfill_candidates = self._apply_backfill_tags_filter_to_results(
+                backfill_candidates, filters.tags
             )
-
-        subquery = db_subquery(self._backfills_query(filters=filters))
-        query = db_select([db.func.count().label("count")]).select_from(subquery)
-        row = self.fetchone(query)
-        count = row["count"] if row else 0
-        return count
+        return len(backfill_candidates)
 
     def get_backfill(self, backfill_id: str) -> PartitionBackfill | None:
         check.str_param(backfill_id, "backfill_id")

python_modules/dagster/dagster_tests/storage_tests/utils/run_storage.py

@@ -1450,14 +1450,6 @@ def test_get_backfills_skips_corrupt_backfill_rows(
             backfill_timestamp=time.time(),
         )
         storage.add_backfill(valid_backfill)
-        storage.add_run(
-            create_dagster_run(
-                run_id=make_new_run_id(),
-                job_name="some_pipeline",
-                status=DagsterRunStatus.SUCCESS,
-                tags={"foo": "bar", BACKFILL_ID_TAG: valid_backfill.backfill_id},
-            )
-        )
 
         with storage.connect() as conn:
             conn.execute(
@@ -1518,6 +1510,41 @@ def test_get_backfills_limit_skips_corrupt_rows_without_shrinking_page(
 
         assert "Skipping backfill corrupt because it could not be deserialized." in caplog.text
 
+    def test_get_backfills_count_skips_corrupt_rows(
+        self, storage: RunStorage, caplog: pytest.LogCaptureFixture
+    ):
+        if not isinstance(storage, SqlRunStorage):
+            pytest.skip("storage is not SQL-backed")
+        if not self.supports_backfills_count():
+            pytest.skip("storage does not support backfill count")
+
+        origin = self.fake_partition_set_origin("fake_partition_set")
+        valid_backfill = PartitionBackfill(
+            "valid",
+            partition_set_origin=origin,
+            status=BulkActionStatus.REQUESTED,
+            partition_names=["a", "b", "c"],
+            from_failure=False,
+            tags={},
+            backfill_timestamp=time.time(),
+        )
+        storage.add_backfill(valid_backfill)
+
+        with storage.connect() as conn:
+            conn.execute(
+                BulkActionsTable.insert().values(
+                    key="corrupt",
+                    status=BulkActionStatus.REQUESTED.value,
+                    timestamp=datetime_from_timestamp(time.time()),
+                    body='{"truncated"',
+                )
+            )
+
+        with caplog.at_level("WARNING", logger="dagster"):
+            assert storage.get_backfills_count() == 1
+
+        assert "Skipping backfill corrupt because it could not be deserialized." in caplog.text
+
     def test_get_backfills_count_with_tags_skips_corrupt_rows(
         self, storage: RunStorage, caplog: pytest.LogCaptureFixture
     ):