[core] swap AssetCheckSummaryRecord for AssetCheckExecutionRecord in asset_graph_view (#24773)

## Summary
- Three call sites in `asset_graph_view.py`
(`compute_subset_with_status`, `_compute_latest_check_run_matches`,
`_compute_updated_since_time_subset`) loaded an
`AssetCheckSummaryRecord` only to read `last_check_execution_record`.
- Swap them to `AssetCheckExecutionRecord.gen`, which routes through
`get_latest_asset_check_execution_by_key` — a single batched query (OSS)
or N queries (cloud) — and returns exactly that record.
- The replaced `get_asset_check_summary_records` path fires N or 2N
queries per call to additionally populate
`last_completed_check_execution_record`, which these callers discard.

## Test plan
- [x] `just ruff`
- [x] `just ty` (0 errors)
- [x] `tox -e py312-asset_defs_tests -- -k asset_graph_view` (66 passed)
- [x] `tox -e py312-declarative_automation_tests -- -k check` (22
passed)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Internal-RevId: 7291b6d9e3ed62901c1d7d3968fe9423b392242e

Author: 2 people

python_modules/dagster-cloud/dagster_cloud/storage/event_logs/queries.py

@@ -123,9 +123,31 @@
     """
 )
 
-ASSET_CHECK_STATE_FRAGMENT = (
+ASSET_CHECK_EXECUTION_RECORD_FRAGMENT = (
     EVENT_LOG_ENTRY_FRAGMENT
     + """
+    fragment AssetCheckExecutionRecordFragment on AssetCheckExecutionRecord {
+        assetCheckKey {
+            assetKey {
+                path
+            }
+            name
+        }
+        id
+        runId
+        status
+        event {
+            ...EventLogEntryFragment
+        }
+        createTimestamp
+        partition
+    }
+"""
+)
+
+ASSET_CHECK_STATE_FRAGMENT = (
+    ASSET_CHECK_EXECUTION_RECORD_FRAGMENT
+    + """
     fragment AssetCheckSummaryRecordFragment on AssetCheckSummaryRecord {
         assetCheckKey {
             assetKey {
@@ -134,22 +156,10 @@
             name
         }
         lastCheckExecutionRecord {
-            id
-            runId
-            status
-            event {
-                ...EventLogEntryFragment
-            }
-            createTimestamp
+            ...AssetCheckExecutionRecordFragment
         }
         lastCompletedCheckExecutionRecord {
-            id
-            runId
-            status
-            event {
-                ...EventLogEntryFragment
-            }
-            createTimestamp
+            ...AssetCheckExecutionRecordFragment
         }
         lastRunId
     }
@@ -297,6 +307,19 @@
 """
 )
 
+GET_LATEST_ASSET_CHECK_EXECUTION_BY_KEY_QUERY = (
+    ASSET_CHECK_EXECUTION_RECORD_FRAGMENT
+    + """
+    query getLatestAssetCheckExecutionByKey($assetCheckKeys: [String!]!, $partitionFilter: PartitionKeyFilter) {
+        eventLogs {
+            getLatestAssetCheckExecutionByKey(assetCheckKeys: $assetCheckKeys, partitionFilter: $partitionFilter) {
+                ...AssetCheckExecutionRecordFragment
+            }
+        }
+    }
+"""
+)
+
 GET_ALL_ASSET_KEYS_QUERY = """
     query getAllAssetKeys {
         eventLogs {

python_modules/dagster-cloud/dagster_cloud/storage/event_logs/storage.py

@@ -93,6 +93,7 @@
     GET_DYNAMIC_PARTITIONS_QUERY,
     GET_EVENT_RECORDS_QUERY,
     GET_EVENT_TAGS_FOR_ASSET,
+    GET_LATEST_ASSET_CHECK_EXECUTION_BY_KEY_QUERY,
     GET_LATEST_ASSET_PARTITION_MATERIALIZATION_ATTEMPTS_WITHOUT_MATERIALIZATIONS,
     GET_LATEST_MATERIALIZATION_EVENTS_QUERY,
     GET_LATEST_PLANNED_MATERIALIZATION_INFO,
@@ -1244,7 +1245,22 @@ def get_latest_asset_check_execution_by_key(
         check_keys: Sequence[AssetCheckKey],
         partition_filter: PartitionKeyFilter | None = None,
     ) -> Mapping[AssetCheckKey, AssetCheckExecutionRecord]:
-        raise NotImplementedError("Not callable from user cloud")
+        if not check_keys:
+            return {}
+        res = self._execute_query(
+            GET_LATEST_ASSET_CHECK_EXECUTION_BY_KEY_QUERY,
+            variables={
+                "assetCheckKeys": [key.to_user_string() for key in check_keys],
+                "partitionFilter": (
+                    {"key": partition_filter.key} if partition_filter is not None else None
+                ),
+            },
+        )
+        records = {}
+        for result in res["data"]["eventLogs"]["getLatestAssetCheckExecutionByKey"]:
+            check_key = AssetCheckKey.from_graphql_input(result["assetCheckKey"])
+            records[check_key] = _asset_check_execution_record_from_graphql(result, check_key)
+        return records
 
     def get_asset_check_partition_info(
         self,

python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py

@@ -585,7 +585,7 @@ async def compute_subset_with_status(
         from_subset: EntitySubset,
     ) -> EntitySubset[AssetCheckKey]:
         """Returns the subset of an asset check that matches a given status."""
-        from dagster._core.storage.event_log.base import AssetCheckSummaryRecord
+        from dagster._core.storage.asset_check_execution_record import AssetCheckExecutionRecord
 
         # Handle partitioned asset checks
         if self._get_partitions_def(key):
@@ -595,8 +595,7 @@ async def compute_subset_with_status(
                 )
 
         # Handle non-partitioned asset checks with existing logic
-        summary = await AssetCheckSummaryRecord.gen(self, key)
-        latest_record = summary.last_check_execution_record if summary else None
+        latest_record = await AssetCheckExecutionRecord.gen(self, key)
         resolved_status = (
             await latest_record.resolve_status(self)
             if latest_record and await latest_record.targets_latest_materialization(self)
@@ -882,12 +881,11 @@ async def _compute_latest_check_run_matches(
         query_key: AssetCheckKey,
         filter_fn: Callable[["RunRecord"], bool],
     ) -> bool:
+        from dagster._core.storage.asset_check_execution_record import AssetCheckExecutionRecord
         from dagster._core.storage.dagster_run import RunRecord
-        from dagster._core.storage.event_log.base import AssetCheckSummaryRecord
 
         check.invariant(partition_key is None, "Partitioned checks not supported")
-        summary = await AssetCheckSummaryRecord.gen(self, query_key)
-        check_record = summary.last_check_execution_record if summary else None
+        check_record = await AssetCheckExecutionRecord.gen(self, query_key)
         if check_record and check_record.event:
             run_record = await RunRecord.gen(self, check_record.event.run_id)
             return bool(run_record) and filter_fn(run_record)
@@ -1011,11 +1009,10 @@ async def _compute_updated_since_time_subset(
         self, key: AssetCheckKey, time: datetime
     ) -> EntitySubset[AssetCheckKey]:
         from dagster._core.events import DagsterEventType
-        from dagster._core.storage.event_log.base import AssetCheckSummaryRecord
+        from dagster._core.storage.asset_check_execution_record import AssetCheckExecutionRecord
 
         # intentionally left unimplemented for AssetKey, as this is a less performant query
-        summary = await AssetCheckSummaryRecord.gen(self, key)
-        record = summary.last_check_execution_record if summary else None
+        record = await AssetCheckExecutionRecord.gen(self, key)
         if (
             record is None
             or record.event is None

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

@@ -5870,6 +5870,9 @@ def test_asset_checks(
         check_key_1 = dg.AssetCheckKey(dg.AssetKey(["my_asset"]), "my_check")
         check_key_2 = dg.AssetCheckKey(dg.AssetKey(["my_asset"]), "my_check_2")
 
+        # empty input short-circuits to empty mapping
+        assert storage.get_latest_asset_check_execution_by_key([]) == {}
+
         for asset_key in {dg.AssetKey(["my_asset"]), dg.AssetKey(["my_other_asset"])}:
             storage.store_event(
                 dg.EventLogEntry(
@@ -5913,6 +5916,13 @@ def test_asset_checks(
         assert latest_checks[check_key_1].status == AssetCheckExecutionRecordStatus.PLANNED
         assert latest_checks[check_key_1].run_id == run_id_1
 
+        # PartitionKeyFilter(key=None) matches unpartitioned records
+        latest_unpartitioned = storage.get_latest_asset_check_execution_by_key(
+            [check_key_1], partition_filter=PartitionKeyFilter(key=None)
+        )
+        assert check_key_1 in latest_unpartitioned
+        assert latest_unpartitioned[check_key_1].partition is None
+
         # update the planned check
         storage.store_event(
             dg.EventLogEntry(
@@ -6038,6 +6048,106 @@ def test_asset_checks(
         latest_checks = storage.get_latest_asset_check_execution_by_key([check_key_1, check_key_2])
         assert len(latest_checks) == 0
 
+    @pytest.mark.parametrize(
+        "partitions_def_type, partition_keys",
+        [
+            ("static", ["a", "b", "c"]),
+            ("dynamic", ["x", "y", "z"]),
+            ("time_window", ["2023-01-01", "2023-01-02", "2023-01-03"]),
+        ],
+        ids=["static", "dynamic", "time_window"],
+    )
+    def test_get_latest_asset_check_execution_by_key_partitioned_lifecycle(
+        self,
+        storage: EventLogStorage,
+        partitions_def_type: str,
+        partition_keys: list,
+    ):
+        run_id = make_new_run_id()
+        asset_key = dg.AssetKey(["my_partitioned_asset_focused"])
+        check_key = dg.AssetCheckKey(asset_key, "my_partitioned_check_focused")
+
+        if partitions_def_type == "static":
+            partitions_def = dg.StaticPartitionsDefinition(partition_keys)
+        elif partitions_def_type == "dynamic":
+            partitions_def = dg.DynamicPartitionsDefinition(
+                name="latest_check_focused_dynamic_partitions"
+            )
+            storage.add_dynamic_partitions(
+                partitions_def_name="latest_check_focused_dynamic_partitions",
+                partition_keys=partition_keys,
+            )
+        elif partitions_def_type == "time_window":
+            partitions_def = dg.DailyPartitionsDefinition(start_date="2023-01-01")
+        else:
+            raise ValueError(f"Unknown partitions_def_type: {partitions_def_type}")
+
+        partitions_subset = partitions_def.subset_with_partition_keys(
+            partition_keys
+        ).to_serializable_subset()
+        key_a, key_b, key_c = partition_keys
+
+        # a single planned event with a multi-partition subset creates one row per partition
+        storage.store_event(
+            _create_check_planned_event(run_id, check_key, partitions_subset=partitions_subset)
+        )
+
+        # each partition is PLANNED in isolation, with the correct partition + run_id
+        for partition_key in partition_keys:
+            result = storage.get_latest_asset_check_execution_by_key(
+                [check_key], partition_filter=PartitionKeyFilter(key=partition_key)
+            )
+            assert check_key in result
+            assert result[check_key].partition == partition_key
+            assert result[check_key].run_id == run_id
+            assert result[check_key].status == AssetCheckExecutionRecordStatus.PLANNED
+
+        # PartitionKeyFilter(key=None) excludes partitioned rows
+        result = storage.get_latest_asset_check_execution_by_key(
+            [check_key], partition_filter=PartitionKeyFilter(key=None)
+        )
+        assert check_key not in result
+
+        # complete partition "a" successfully
+        storage.store_event(
+            _create_check_evaluation_event(run_id, check_key, passed=True, partition=key_a)
+        )
+
+        # "a" is SUCCEEDED; "b" and "c" still PLANNED
+        assert (
+            storage.get_latest_asset_check_execution_by_key(
+                [check_key], partition_filter=PartitionKeyFilter(key=key_a)
+            )[check_key].status
+            == AssetCheckExecutionRecordStatus.SUCCEEDED
+        )
+        for partition_key in (key_b, key_c):
+            assert (
+                storage.get_latest_asset_check_execution_by_key(
+                    [check_key], partition_filter=PartitionKeyFilter(key=partition_key)
+                )[check_key].status
+                == AssetCheckExecutionRecordStatus.PLANNED
+            )
+
+        # fail partition "b"
+        storage.store_event(
+            _create_check_evaluation_event(run_id, check_key, passed=False, partition=key_b)
+        )
+
+        expected = {
+            key_a: AssetCheckExecutionRecordStatus.SUCCEEDED,
+            key_b: AssetCheckExecutionRecordStatus.FAILED,
+            key_c: AssetCheckExecutionRecordStatus.PLANNED,
+        }
+        for partition_key, status in expected.items():
+            record = storage.get_latest_asset_check_execution_by_key(
+                [check_key], partition_filter=PartitionKeyFilter(key=partition_key)
+            )[check_key]
+            assert record.partition == partition_key
+            assert record.status == status
+
+        latest = storage.get_latest_asset_check_execution_by_key([check_key])
+        assert check_key in latest
+
     def test_duplicate_asset_check_planned_events(self, storage: EventLogStorage):
         run_id = make_new_run_id()
         for _ in range(2):