fix

Author: jamiedemaria

js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql

@@ -1,4 +1,5 @@
 import asyncio
+from typing import Optional
 
 import graphene
 from dagster._core.definitions.asset_health.asset_check_health import (
@@ -22,6 +23,7 @@
     AssetHealthMaterializationMetadata,
     get_materialization_status_and_metadata,
 )
+from dagster._core.storage.event_log.base import AssetRecord
 
 from dagster_graphql.schema.entity_key import GrapheneAssetKey
 from dagster_graphql.schema.util import ResolveInfo
@@ -91,11 +93,38 @@ class GrapheneAssetHealthMaterializationDegradedPartitionedMeta(graphene.ObjectT
     numFailedPartitions = graphene.NonNull(graphene.Int)
     numMissingPartitions = graphene.NonNull(graphene.Int)
     totalNumPartitions = graphene.NonNull(graphene.Int)
+    latestRunId = graphene.String()
     failedRunId = graphene.String()
 
     class Meta:
         name = "AssetHealthMaterializationDegradedPartitionedMeta"
 
+    def __init__(
+        self,
+        asset_key: GrapheneAssetKey,
+        num_failed_partitions: int,
+        num_missing_partitions: int,
+        total_num_partitions: int,
+        latest_run_id: str,
+    ):
+        super().__init__()
+        self._asset_key = asset_key
+        self.numFailedPartitions = num_failed_partitions
+        self.numMissingPartitions = num_missing_partitions
+        self.totalNumPartitions = total_num_partitions
+        self.latestRunId = latest_run_id
+
+    async def resolve_failedRunId(self, graphene_info: ResolveInfo) -> Optional[str]:
+        # The latest failed run for a degraded partitioned asset may not be the most recent run that
+        # attempted to materialize the asset. The failed partitions could be in a historical run
+        asset_record = await AssetRecord.gen(graphene_info.context, self._asset_key)
+        if (
+            asset_record is not None
+            and asset_record.asset_entry.last_failed_to_materialize_record is not None
+        ):
+            return asset_record.asset_entry.last_failed_to_materialize_record.run_id
+        return None
+
 
 class GrapheneAssetHealthMaterializationHealthyPartitionedMeta(graphene.ObjectType):
     numMissingPartitions = graphene.NonNull(graphene.Int)
@@ -124,6 +153,7 @@ class Meta:
 
     @staticmethod
     def from_metadata_class(
+        asset_key: GrapheneAssetKey,
         metadata: AssetHealthMaterializationMetadata,
     ) -> "GrapheneAssetHealthMaterializationMeta":
         if isinstance(metadata, AssetHealthMaterializationDegradedNotPartitionedMeta):
@@ -138,10 +168,11 @@ def from_metadata_class(
             )
         elif isinstance(metadata, AssetHealthMaterializationDegradedPartitionedMeta):
             return GrapheneAssetHealthMaterializationDegradedPartitionedMeta(
-                numFailedPartitions=metadata.num_failed_partitions,
-                numMissingPartitions=metadata.num_missing_partitions,
-                totalNumPartitions=metadata.total_num_partitions,
-                failedRunId=metadata.failed_run_id,
+                asset_key=asset_key,
+                num_failed_partitions=metadata.num_failed_partitions,
+                num_missing_partitions=metadata.num_missing_partitions,
+                total_num_partitions=metadata.total_num_partitions,
+                latest_run_id=metadata.latest_run_id,
             )
         else:
             raise ValueError(f"Unknown metadata class: {type(metadata)}")
@@ -192,7 +223,7 @@ async def resolve_materializationStatusMetadata(
         _, materialization_status_metadata = await self.materialization_status_task
         return (
             GrapheneAssetHealthMaterializationMeta.from_metadata_class(
-                materialization_status_metadata
+                asset_key=self._asset_key, metadata=materialization_status_metadata
             )
             if materialization_status_metadata
             else None

js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts

@@ -331,7 +331,7 @@ class AssetHealthMaterializationDegradedPartitionedMeta:
     num_failed_partitions: int
     num_missing_partitions: int
     total_num_partitions: int
-    failed_run_id: Optional[str]
+    latest_run_id: Optional[str]
 
 
 @whitelist_for_serdes
@@ -462,7 +462,7 @@ async def get_materialization_status_and_metadata(
                 num_failed_partitions=asset_materialization_health_state.num_failed_partitions,
                 num_missing_partitions=num_missing,
                 total_num_partitions=total_num_partitions,
-                failed_run_id=asset_materialization_health_state.latest_terminal_run_id,
+                latest_run_id=asset_materialization_health_state.latest_terminal_run_id,
             )
         else:
             meta = AssetHealthMaterializationDegradedNotPartitionedMeta(