start process of removing tags when a run is retried from a completed…

… backfill
dagster-io · Nov 13, 2024 · 039026a · 039026a
1 parent 46e48b3
commit 039026a
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 5 deletions.
diff --git a/python_modules/dagster/dagster/_core/instance/__init__.py b/python_modules/dagster/dagster/_core/instance/__init__.py
@@ -70,10 +70,13 @@
 from dagster._core.storage.tags import (
     ASSET_PARTITION_RANGE_END_TAG,
     ASSET_PARTITION_RANGE_START_TAG,
+    BACKFILL_ID_TAG,
+    BACKFILL_TAGS,
     PARENT_RUN_ID_TAG,
     PARTITION_NAME_TAG,
     RESUME_RETRY_TAG,
     ROOT_RUN_ID_TAG,
+    TAGS_TO_MAYBE_OMIT_ON_RETRY,
     TAGS_TO_OMIT_ON_RETRY,
 )
 from dagster._serdes import ConfigurableClass
@@ -1627,6 +1630,7 @@ def create_reexecuted_run(
         run_config: Optional[Mapping[str, Any]] = None,
         use_parent_run_tags: bool = False,
     ) -> DagsterRun:
+        from dagster._core.execution.backfill import BulkActionStatus
         from dagster._core.execution.plan.resume_retry import ReexecutionStrategy
         from dagster._core.execution.plan.state import KnownExecutionState
         from dagster._core.remote_representation import CodeLocation, RemoteJob
@@ -1644,11 +1648,21 @@ def create_reexecuted_run(
         parent_run_id = parent_run.run_id
 
         # these can differ from remote_job.tags if tags were added at launch time
-        parent_run_tags = (
-            {key: val for key, val in parent_run.tags.items() if key not in TAGS_TO_OMIT_ON_RETRY}
-            if use_parent_run_tags
-            else {}
-        )
+        parent_run_tags = {}
+        if use_parent_run_tags:
+            parent_run_tags = {
+                key: val
+                for key, val in parent_run.tags.items()
+                if key not in TAGS_TO_OMIT_ON_RETRY and key not in TAGS_TO_MAYBE_OMIT_ON_RETRY
+            }
+            # if the run was part of a backfill and the backfill is complete, we do not want the
+            # retry to be considered part of the backfill, so remove all backfill-related tags
+            if parent_run.tags.get(BACKFILL_ID_TAG) is not None:
+                backfill = self.get_backfill(parent_run.tags[BACKFILL_ID_TAG])
+                if backfill.status == BulkActionStatus.REQUESTED:
+                    for tag in BACKFILL_TAGS:
+                        if parent_run.tags.get(tag) is not None:
+                            parent_run_tags[tag] = parent_run.tags[tag]
 
         tags = merge_dicts(
             remote_job.tags,

diff --git a/python_modules/dagster/dagster/_core/storage/tags.py b/python_modules/dagster/dagster/_core/storage/tags.py
@@ -101,9 +101,13 @@
 RUN_METRICS_POLLING_INTERVAL_TAG = f"{HIDDEN_TAG_PREFIX}run_metrics_polling_interval"
 RUN_METRICS_PYTHON_RUNTIME_TAG = f"{HIDDEN_TAG_PREFIX}python_runtime_metrics"
 
+BACKFILL_TAGS = {BACKFILL_ID_TAG, PARENT_BACKFILL_ID_TAG, ROOT_BACKFILL_ID_TAG}
+
 
 TAGS_TO_OMIT_ON_RETRY = {*RUN_METRIC_TAGS, RUN_FAILURE_REASON_TAG}
 
+TAGS_TO_MAYBE_OMIT_ON_RETRY = {*BACKFILL_TAGS}
+
 
 class TagType(Enum):
     # Custom tag provided by a user