Pass dynamic_partitions_store to has_partition_key for DynamicPartitionsDefinition and MultiPartitionsDefinition

Fixed issue where DynamicPartitionsDefinition.has_partition_key was being
called without the required dynamic_partitions_store parameter, causing
tests to fail with "The instance is not available to load partitions" error.

Changes:
1. Updated MultiPartitionsDefinition.has_partition_key to accept current_time
   and dynamic_partitions_store parameters, passing them through to
   DynamicPartitionsDefinition dimensions

2. Updated run_domain._get_planned_partitions to pass dynamic_partitions_store
   when checking has_partition_key for both DynamicPartitionsDefinition and
   MultiPartitionsDefinition

This ensures dynamic partitions can access the instance to load partition keys,
whether they're used directly or as dimensions in multi-partitions.

Fixes: test_dynamic_partitioned_run, test_unfetched_partitioned_events_are_unconsumed,
       test_assets (dynamic_asset_partitions)

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

Co-Authored-By: Claude <noreply@anthropic.com>

Author: prha

python_modules/dagster/dagster/_core/definitions/partitions/definition/multi.py

@@ -190,7 +190,16 @@ def get_partitions_def_for_dimension(self, dimension_name: str) -> PartitionsDef
         check.failed(f"Invalid dimension name {dimension_name}")
 
     # We override the default implementation of `has_partition_key` for performance.
-    def has_partition_key(self, partition_key: Union[MultiPartitionKey, str]) -> bool:
+    def has_partition_key(
+        self,
+        partition_key: Union[MultiPartitionKey, str],
+        current_time: Optional[datetime] = None,
+        dynamic_partitions_store: Optional["DynamicPartitionsStore"] = None,
+    ) -> bool:
+        from dagster._core.definitions.partitions.definition.dynamic import (
+            DynamicPartitionsDefinition,
+        )
+
         if isinstance(partition_key, str):
             try:
                 partition_key = self.get_partition_key_from_str(partition_key)
@@ -204,9 +213,19 @@ def has_partition_key(self, partition_key: Union[MultiPartitionKey, str]) -> boo
             )
 
         for dimension in self.partitions_defs:
-            if not dimension.partitions_def.has_partition_key(
-                partition_key.keys_by_dimension[dimension.name]
-            ):
+            # For DynamicPartitionsDefinition, pass the dynamic_partitions_store
+            if isinstance(dimension.partitions_def, DynamicPartitionsDefinition):
+                has_key = dimension.partitions_def.has_partition_key(
+                    partition_key.keys_by_dimension[dimension.name],
+                    current_time=current_time,
+                    dynamic_partitions_store=dynamic_partitions_store,
+                )
+            else:
+                has_key = dimension.partitions_def.has_partition_key(
+                    partition_key.keys_by_dimension[dimension.name]
+                )
+
+            if not has_key:
                 return False
         return True
 

python_modules/dagster/dagster/_core/instance/runs/run_domain.py

@@ -954,7 +954,20 @@ def _get_planned_partitions(
                 f" {PARTITION_NAME_TAG}"
             )
 
-        partitions_def = asset_graph.get(asset_key_or_check_key).partitions_def
+        # Get the node from the asset graph - handle both AssetKey and AssetCheckKey
+        if isinstance(asset_key_or_check_key, AssetKey):
+            node = self._get_repo_scoped_asset_node(
+                asset_graph, asset_key_or_check_key, dagster_run.remote_job_origin
+            )
+        else:
+            # For AssetCheckKey, check if it exists before trying to get it
+            node = (
+                asset_graph.get(asset_key_or_check_key)
+                if asset_graph.has(asset_key_or_check_key)
+                else None
+            )
+
+        partitions_def = node.partitions_def if node else None
         partitions_subset = None
         individual_partitions = None
         if partition_range_start or partition_range_end:
@@ -985,8 +998,13 @@ def _get_planned_partitions(
                             PartitionKeyRange(partition_range_start, partition_range_end),
                         )
         elif partitions_def and partition_tag:
-            # For DynamicPartitionsDefinition, we need to pass the instance to check partition existence
-            if isinstance(partitions_def, DynamicPartitionsDefinition):
+            # For DynamicPartitionsDefinition and MultiPartitionsDefinition (which may contain dynamic dimensions),
+            # we need to pass the instance to check partition existence
+            from dagster._core.definitions.partitions.definition.multi import (
+                MultiPartitionsDefinition,
+            )
+
+            if isinstance(partitions_def, (DynamicPartitionsDefinition, MultiPartitionsDefinition)):
                 has_key = partitions_def.has_partition_key(
                     partition_tag, dynamic_partitions_store=self._instance
                 )