Repository: dbt-labs/dbt-core (python)

Issue:
[Bug] dbt retry should use the original failure date as reference for microbatch models

### Is this a new bug in dbt-core?

- [x] I believe this is a new bug in dbt-core
- [x] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

When a microbatch model fails during `dbt run`, and we later execute `dbt retry`, the command uses the current date as reference instead of the original failure date. This results in processing different batches than those that originally failed.


### Expected Behavior

When executing `dbt retry`, it should use the original failure date (2025-03-21 in this case) as the reference date for microbatch processing, retrying the batches that originally failed (2025-03-18 through 2025-03-21).

### Steps To Reproduce

### 1. On 2025-03-21, executed the following microbatch model
**Command:**
```
dbt run --target dev --log-level info --target-path /target-path/job --select target_table
```

**Model configuration:**
```
{{
    config(
        materialized='incremental',
        incremental_strategy='microbatch',
        event_time='_event_date',
        begin='2024-12-01',
        lookback=3,
        batch_size='day'
    )
}}
```

**Result:**
```
07:55:33  1 of 8 START sql microbatch model schema.target_table ................. [RUN]
07:55:33  Batch 1 of 4 START batch 2025-03-18 of schema.target_table .................. [RUN]
07:55:35  Batch 1 of 4 ERROR creating batch 2025-03-18 of schema.target_table ......... [ERROR in 2.56s]
07:55:35  Batch 2 of 4 SKIP batch 2025-03-19 of schema.target_table ................... [SKIPPED in 0.00s]
07:55:35  Batch 3 of 4 SKIP batch 2025-03-20 of schema.target_table ................... [SKIPPED in 0.00s]
07:55:35  Batch 4 of 4 SKIP batch 2025-03-21 of schema.target_table ................... [SKIPPED in 0.00s]
07:55:35  1 of 8 ERROR creating sql microbatch model schema.target_table ........ [ERROR in 2.61s]
```

### 2. On 2025-03-25, executed dbt retry
**Command:**
```
dbt retry --target dev --log-level info --state /target-path/job
```

**Result:**
```
01:04:21  1 of 1 START sql microbatch model schema.target_table ................. [RUN]
01:04:21  Batch 1 of 4 START batch 2025-03-22 of schema.target_table .................. [RUN]
01:04:30  Batch 1 of 4 OK created batch 2025-03-22 of schema.target_table ............. [OK in 9.19s]
01:04:30  Batch 2 of 4 START batch 2025-03-23 of schema.target_table .................. [RUN]
01:04:37  Batch 2 of 4 OK created batch 2025-03-23 of schema.target_table ............. [OK in 6.75s]
01:04:37  Batch 3 of 4 START batch 2025-03-24 of schema.target_table .................. [RUN]
01:04:44  Batch 3 of 4 OK created batch 2025-03-24 of schema.target_table ............. [OK in 6.53s]
01:04:44  Batch 4 of 4 START batch 2025-03-25 of schema.target_table .................. [RUN]
01:04:50  Batch 4 of 4 OK created batch 2025-03-25 of schema.target_table ............. [OK in 6.25s]
01:04:50  1 of 1 OK created sql microbatch model schema.target_table ............ [SUCCESS in 28.78s]
```

### Relevant log output

```shell

```

### Environment

```markdown
- OS:Ubuntu24.04
- Python:3.12.0
- dbt:1.9.3
```

### Which database adapter are you using with dbt?

_No response_

### Additional Context

The condition here is suspicious.
https://github.com/dbt-labs/dbt-core/blob/v1.9.3/core/dbt/task/retry.py#L144


Patch:
diff --git a/.changes/unreleased/Fixes-20260223-161820.yaml b/.changes/unreleased/Fixes-20260223-161820.yaml
new file mode 100644
index 00000000000..03e3cd017e1
--- /dev/null
+++ b/.changes/unreleased/Fixes-20260223-161820.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix dbt retry for microbatch models to use the original invocation time instead of the current time when recomputing batches
+time: 2026-02-23T16:18:20.714612+05:30
+custom:
+    Author: aahel
+    Issue: "11423"
diff --git a/core/dbt/task/retry.py b/core/dbt/task/retry.py
index 9f0b0ecd9f8..654eaa2d823 100644
--- a/core/dbt/task/retry.py
+++ b/core/dbt/task/retry.py
@@ -171,8 +171,14 @@ def get_graph_queue(self):
             self.manifest,
         )
 
-        if self.task_class == RunTask:
+        # issubclass ensures that BuildTask (which extends RunTask)
+        # also gets microbatch retry behavior
+        if issubclass(self.task_class, RunTask):
             task.batch_map = batch_map
+            task.original_invocation_started_at = (
+                self.previous_results.metadata.invocation_started_at
+                or self.previous_results.metadata.generated_at
+            )
 
         return_value = task.run()
         return return_value
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 24038010c77..f093abef55b 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -727,12 +727,18 @@ def get_microbatch_builder(self, model: ModelNode) -> MicrobatchBuilder:
             event_time_start = self.config.args.sample.start
             event_time_end = self.config.args.sample.end
 
+        # During retry, use the original invocation time so that the same
+        # batches are recomputed rather than batches relative to "now".
+        default_end_time = (
+            self.parent_task.original_invocation_started_at or get_invocation_started_at()
+        )
+
         return MicrobatchBuilder(
             model=model,
             is_incremental=self._is_incremental(model),
             event_time_start=event_time_start,
             event_time_end=event_time_end,
-            default_end_time=get_invocation_started_at(),
+            default_end_time=default_end_time,
         )
 
     def get_batches(self, model: ModelNode) -> Dict[int, BatchType]:
@@ -846,6 +852,7 @@ def __init__(
     ) -> None:
         super().__init__(args, config, manifest)
         self.batch_map = batch_map
+        self.original_invocation_started_at: Optional[datetime] = None
 
     def raise_on_first_error(self) -> bool:
         return False


Test patch:
diff --git a/tests/functional/microbatch/test_microbatch.py b/tests/functional/microbatch/test_microbatch.py
index 38c6c0a2810..9ed31caefb0 100644
--- a/tests/functional/microbatch/test_microbatch.py
+++ b/tests/functional/microbatch/test_microbatch.py
@@ -1,4 +1,5 @@
 import json
+import os
 from typing import Dict
 from unittest import mock
 
@@ -1296,3 +1297,105 @@ def macros(self) -> Dict[str, str]:
     def test_microbatch(self, project) -> None:
         _, console_output = run_dbt_and_capture(["run"], expect_pass=False)
         assert "Completed with 1 error, 0 partial successes, and 0 warnings" in console_output
+
+
+microbatch_model_all_failing_sql = """
+{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='day', begin=modules.datetime.datetime(2020, 1, 1, 0, 0, 0)) }}
+ invalid_sql
+select * from {{ ref('input_model') }}
+"""
+
+
+class TestMicrobatchRetryUsesOriginalInvocationTime(BaseMicrobatchTest):
+    """When all batches fail and dbt retry is run later, the retry should
+    recompute batches using the original invocation time, not the current time."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_all_failing_sql,
+        }
+
+    def test_retry_uses_original_invocation_time(self, project):
+        # Run with end time of 2020-01-03 — produces 3 batches, all fail
+        with patch_microbatch_end_time("2020-01-03 13:57:00"):
+            run_dbt(["run"], expect_pass=False)
+
+        run_results = get_artifact(project.project_root, "target", "run_results.json")
+        microbatch_result = next(
+            r for r in run_results["results"] if "microbatch_model" in r["unique_id"]
+        )
+        assert microbatch_result["status"] == "error"
+
+        # Simulate the original run happening at 2020-01-03 by updating
+        # invocation_started_at in the saved run_results.json
+        run_results_path = os.path.join(project.project_root, "target", "run_results.json")
+        run_results["metadata"]["invocation_started_at"] = "2020-01-03T13:57:00Z"
+        with open(run_results_path, "w") as f:
+            json.dump(run_results, f)
+
+        # Fix the model so retry can succeed
+        write_file(
+            microbatch_model_sql,
+            project.project_root,
+            "models",
+            "microbatch_model.sql",
+        )
+
+        # Run retry WITHOUT patching end time. The fix ensures retry
+        # uses the original invocation time from run_results.json (2020-01-03),
+        # so only 3 batches are produced. Without the fix, it would use
+        # the current time and produce many more batches.
+        batch_catcher = EventCatcher(event_to_catch=LogBatchResult)
+        run_dbt(["retry"], callbacks=[batch_catcher.catch])
+
+        # Should only have 3 batches (2020-01-01, 2020-01-02, 2020-01-03)
+        assert len(batch_catcher.caught_events) == 3
+        self.assert_row_count(project, "microbatch_model", 3)
+
+
+class TestMicrobatchBuildRetryUsesOriginalInvocationTime(BaseMicrobatchTest):
+    """Same as above but using dbt build + retry to verify the issubclass
+    check ensures BuildTask also gets original invocation time behavior."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_all_failing_sql,
+        }
+
+    def test_build_retry_uses_original_invocation_time(self, project):
+        # Run build with end time of 2020-01-03 — produces 3 batches, all fail
+        with patch_microbatch_end_time("2020-01-03 13:57:00"):
+            run_dbt(["build"], expect_pass=False)
+
+        run_results = get_artifact(project.project_root, "target", "run_results.json")
+        microbatch_result = next(
+            r for r in run_results["results"] if "microbatch_model" in r["unique_id"]
+        )
+        assert microbatch_result["status"] == "error"
+
+        # Simulate the original build happening at 2020-01-03 by updating
+        # invocation_started_at in the saved run_results.json
+        run_results_path = os.path.join(project.project_root, "target", "run_results.json")
+        run_results["metadata"]["invocation_started_at"] = "2020-01-03T13:57:00Z"
+        with open(run_results_path, "w") as f:
+            json.dump(run_results, f)
+
+        # Fix the model so retry can succeed
+        write_file(
+            microbatch_model_sql,
+            project.project_root,
+            "models",
+            "microbatch_model.sql",
+        )
+
+        # Retry should use the original invocation time, not "now"
+        batch_catcher = EventCatcher(event_to_catch=LogBatchResult)
+        run_dbt(["retry"], callbacks=[batch_catcher.catch])
+
+        # Should only have 3 batches (2020-01-01, 2020-01-02, 2020-01-03)
+        assert len(batch_catcher.caught_events) == 3
+        self.assert_row_count(project, "microbatch_model", 3)
diff --git a/tests/unit/task/test_run.py b/tests/unit/task/test_run.py
index 3e2c7605ef3..77827d8bd65 100644
--- a/tests/unit/task/test_run.py
+++ b/tests/unit/task/test_run.py
@@ -1,5 +1,6 @@
 from argparse import Namespace
 from dataclasses import dataclass
+from datetime import datetime
 from importlib import import_module
 from typing import Optional, Type, Union
 from unittest import mock
@@ -304,6 +305,78 @@ def test_should_run_in_parallel(
         # Assert result of should_run_in_parallel
         assert batch_runner.should_run_in_parallel() == expectation
 
+    def test_get_microbatch_builder_uses_original_invocation_time_on_retry(
+        self,
+        mocker: MockerFixture,
+        model_runner: MicrobatchModelRunner,
+    ) -> None:
+        """When retrying, get_microbatch_builder should use the original invocation
+        time from the previous run rather than the current invocation time."""
+        original_time = datetime(2025, 3, 21, 7, 55, 0)
+        current_time = datetime(2025, 3, 25, 1, 4, 0)
+
+        # Set up a mock parent task with original_invocation_started_at
+        mock_parent = mocker.Mock(spec=RunTask)
+        mock_parent.original_invocation_started_at = original_time
+        model_runner._parent_task = mock_parent
+
+        # Mock _is_incremental to avoid adapter calls
+        mocker.patch.object(model_runner, "_is_incremental", return_value=True)
+
+        # Mock get_invocation_started_at to return the "current" (retry) time
+        mocker.patch(
+            "dbt.task.run.get_invocation_started_at",
+            return_value=current_time,
+        )
+
+        model = model_runner.node
+        model.config.materialized = "incremental"
+        model.config.incremental_strategy = "microbatch"
+        model.config.batch_size = "day"
+        model.config.begin = "2024-12-01"
+        model.config.event_time = "_event_date"
+        model_runner.config.args = Namespace(
+            EVENT_TIME_START=None, EVENT_TIME_END=None, SAMPLE=None
+        )
+
+        builder = model_runner.get_microbatch_builder(model)
+        assert builder.default_end_time == original_time
+
+    def test_get_microbatch_builder_uses_current_time_without_retry(
+        self,
+        mocker: MockerFixture,
+        model_runner: MicrobatchModelRunner,
+    ) -> None:
+        """When not retrying (normal run), get_microbatch_builder should use
+        the current invocation time."""
+        current_time = datetime(2025, 3, 25, 1, 4, 0)
+
+        # Set up a mock parent task with no original_invocation_started_at
+        mock_parent = mocker.Mock(spec=RunTask)
+        mock_parent.original_invocation_started_at = None
+        model_runner._parent_task = mock_parent
+
+        # Mock _is_incremental to avoid adapter calls
+        mocker.patch.object(model_runner, "_is_incremental", return_value=True)
+
+        mocker.patch(
+            "dbt.task.run.get_invocation_started_at",
+            return_value=current_time,
+        )
+
+        model = model_runner.node
+        model.config.materialized = "incremental"
+        model.config.incremental_strategy = "microbatch"
+        model.config.batch_size = "day"
+        model.config.begin = "2024-12-01"
+        model.config.event_time = "_event_date"
+        model_runner.config.args = Namespace(
+            EVENT_TIME_START=None, EVENT_TIME_END=None, SAMPLE=None
+        )
+
+        builder = model_runner.get_microbatch_builder(model)
+        assert builder.default_end_time == current_time
+
 
 class TestRunTask:
     @pytest.fixture