testing

dagster-io · Aug 15, 2024 · b927649 · b927649
1 parent e48bc24
commit b927649
Show file tree

Hide file tree

Showing 6 changed files with 500 additions and 8 deletions.
diff --git a/js_modules/dagster-ui/packages/ui-core/src/graphql/possibleTypes.generated.json b/js_modules/dagster-ui/packages/ui-core/src/graphql/possibleTypes.generated.json
diff --git a/js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql b/js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql
diff --git a/js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts b/js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts
diff --git a/python_modules/dagster-graphql/dagster_graphql/implementation/execution/backfill.py b/python_modules/dagster-graphql/dagster_graphql/implementation/execution/backfill.py
@@ -10,7 +10,11 @@
 )
 from dagster._core.events import AssetKey
 from dagster._core.execution.asset_backfill import create_asset_backfill_data_from_asset_partitions
-from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+from dagster._core.execution.backfill import (
+    BULK_ACTION_COMPLETED_STATUSES,
+    BulkActionStatus,
+    PartitionBackfill,
+)
 from dagster._core.execution.job_backfill import submit_backfill_runs
 from dagster._core.remote_representation.external_data import ExternalPartitionExecutionErrorData
 from dagster._core.storage.tags import PARENT_RUN_ID_TAG, ROOT_RUN_ID_TAG
@@ -331,13 +335,27 @@ def retry_partition_backfill(
     if not backfill:
         check.failed(f"No backfill found for id: {backfill_id}")
 
-    partition_set_origin = check.not_none(backfill.partition_set_origin)
-    location_name = partition_set_origin.selector.location_name
-    assert_permission_for_location(
-        graphene_info, Permissions.LAUNCH_PARTITION_BACKFILL, location_name
-    )
+    if backfill.status not in BULK_ACTION_COMPLETED_STATUSES:
+        raise DagsterInvariantViolationError(
+            f"Cannot retry backfill {backfill_id} because it is still in progress."
+        )
 
     if backfill.is_asset_backfill:
+        if (
+            backfill.asset_backfill_data.failed_and_downstream_subset.num_partitions_and_non_partitioned_assets
+            == 0
+        ):
+            raise DagsterInvariantViolationError(
+                "Cannot retry an asset backfill that has no failed assets."
+            )
+        asset_graph = graphene_info.context.asset_graph
+        assert_permission_for_asset_graph(
+            graphene_info,
+            asset_graph,
+            backfill.asset_backfill_data.failed_and_downstream_subset.asset_keys,
+            Permissions.LAUNCH_PARTITION_BACKFILL,
+        )
+
         new_backfill = PartitionBackfill.from_asset_graph_subset(
             backfill_id=make_new_backfill_id(),
             asset_graph_subset=backfill.asset_backfill_data.failed_and_downstream_subset,
@@ -352,8 +370,14 @@ def retry_partition_backfill(
             description=backfill.description,
         )
     else:
+        partition_set_origin = check.not_none(backfill.partition_set_origin)
+        location_name = partition_set_origin.selector.location_name
+        assert_permission_for_location(
+            graphene_info, Permissions.LAUNCH_PARTITION_BACKFILL, location_name
+        )
+
         new_backfill = PartitionBackfill(
-            backfill_id=backfill.backfill_id,
+            backfill_id=make_new_backfill_id(),
             partition_set_origin=backfill.partition_set_origin,
             status=BulkActionStatus.REQUESTED,
             partition_names=backfill.partition_names,