Repository: dbt-labs/dbt-core
Pull request: Ensure consistent `current_time` across microbatch models in an invocation
Resolves #10819

### Problem

Different microbatch models in the same `dbt run` could end up with different `current_time`. This could cause a situation where two microbatch models operating on the same inputs could end up with more/less data if one microbatch model was executed significantly later than the other.

### Solution

Make it so each microbatch model in an invocation is using the same `current_time`

### Checklist

- [X] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me.
- [X] I have run this code in development, and it appears to resolve the stated issue.
- [X] This PR includes tests, or tests are not required or relevant for this PR.
- [X] This PR has no interface changes (e.g., macros, CLI, logs, JSON artifacts, config files, adapter interface, etc.) or this PR has already received feedback and approval from Product or DX.
- [X] This PR includes [type annotations](https://docs.python.org/3/library/typing.html) for new and modified functions.


Problem statement:
Set a singular "current_time" for microbatch models per invocation
Currently, microbatch models get a "current_time" (`datetime.datetime.now(pytz.UTC)`) when they are executed. Notably, each microbatch model gets a _different_ "current_time". This works, but has some funkiness.

Consider the following:
1. There is a source, `source_1`, which is constantly being updated by an external process
2. Microbatch model `model_a` pulls from `source_1`
3. Microbatch model `model_b` pulls from `source_1`

Regardless if we're in a small or large project, given _any_ delay in the execution of `model_a` and `model_b`, if there is any new data in `source_1`, the result of `model_a` will be _different_ from `model_b`. An example would be:

1. `source_1` has 3 rows with event times: `2024-10-07 12:17:00`, `2024-10-07 12:16:00`, `2024-10-07 12:15:00`
2. the "current_time" is `2024-10-07 12:18:00`
3. `model_a` is executed, picking up the 3 rows
4. new data arrives to `source_1`: `2024-10-07 12:18:30`
5. the "current_time" is `2024-10-07 12:19:00`
6. `model_b` is executed, picking up 4 rows.

Is the discrepancy okay?

Hints from issue discussion:

Commit message: Add changie doc

Patch to review:
diff --git a/.changes/unreleased/Features-20241007-115853.yaml b/.changes/unreleased/Features-20241007-115853.yaml
new file mode 100644
index 00000000000..ac2e61c5b59
--- /dev/null
+++ b/.changes/unreleased/Features-20241007-115853.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Ensure microbatch models use same `current_time` value
+time: 2024-10-07T11:58:53.460941-05:00
+custom:
+  Author: QMalcolm
+  Issue: "10819"
diff --git a/core/dbt/config/runtime.py b/core/dbt/config/runtime.py
index e1c24cf5f0c..87747f5462a 100644
--- a/core/dbt/config/runtime.py
+++ b/core/dbt/config/runtime.py
@@ -1,7 +1,8 @@
 import itertools
 import os
 from copy import deepcopy
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from datetime import datetime
 from pathlib import Path
 from typing import (
     Any,
@@ -15,6 +16,8 @@
     Type,
 )
 
+import pytz
+
 from dbt import tracking
 from dbt.adapters.contracts.connection import (
     AdapterRequiredConfig,
@@ -98,6 +101,7 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
     profile_name: str
     cli_vars: Dict[str, Any]
     dependencies: Optional[Mapping[str, "RuntimeConfig"]] = None
+    invocated_at: datetime = field(default_factory=lambda: datetime.now(pytz.UTC))
 
     def __post_init__(self):
         self.validate()
diff --git a/core/dbt/materializations/incremental/microbatch.py b/core/dbt/materializations/incremental/microbatch.py
index 752793e124b..439e07204b1 100644
--- a/core/dbt/materializations/incremental/microbatch.py
+++ b/core/dbt/materializations/incremental/microbatch.py
@@ -18,6 +18,7 @@ def __init__(
         is_incremental: bool,
         event_time_start: Optional[datetime],
         event_time_end: Optional[datetime],
+        batch_current_time: Optional[datetime] = None,
     ):
         if model.config.incremental_strategy != "microbatch":
             raise DbtInternalError(
@@ -35,10 +36,13 @@ def __init__(
             event_time_start.replace(tzinfo=pytz.UTC) if event_time_start else None
         )
         self.event_time_end = event_time_end.replace(tzinfo=pytz.UTC) if event_time_end else None
+        self.batch_current_time = (
+            batch_current_time if batch_current_time is not None else datetime.now(pytz.UTC)
+        )
 
     def build_end_time(self):
         """Defaults the end_time to the current time in UTC unless a non `None` event_time_end was provided"""
-        return self.event_time_end or datetime.now(tz=pytz.utc)
+        return self.event_time_end or self.batch_current_time
 
     def build_start_time(self, checkpoint: Optional[datetime]):
         """Create a start time based off the passed in checkpoint.
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 93a8e261b50..9e084330125 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -487,6 +487,7 @@ def _execute_microbatch_materialization(
                 is_incremental=self._is_incremental(model),
                 event_time_start=getattr(self.config.args, "EVENT_TIME_START", None),
                 event_time_end=getattr(self.config.args, "EVENT_TIME_END", None),
+                batch_current_time=self.config.invocated_at,
             )
             end = microbatch_builder.build_end_time()
             start = microbatch_builder.build_start_time(end)
diff --git a/tests/functional/microbatch/test_microbatch.py b/tests/functional/microbatch/test_microbatch.py
index c867acfac66..5ffbb975e6c 100644
--- a/tests/functional/microbatch/test_microbatch.py
+++ b/tests/functional/microbatch/test_microbatch.py
@@ -1,7 +1,9 @@
 import os
+from datetime import datetime
 from unittest import mock
 
 import pytest
+import pytz
 
 from dbt.tests.util import (
     get_artifact,
@@ -38,6 +40,11 @@
 select * from {{ ref('input_model') }}
 """
 
+microbatch_model_downstream_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)) }}
+select * from {{ ref('microbatch_model') }}
+"""
+
 
 microbatch_model_ref_render_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)) }}
@@ -657,3 +664,28 @@ def test_run_with_event_time(self, project):
         with patch_microbatch_end_time("2020-01-03 13:57:00"):
             run_dbt(["run", "--full-refresh"])
         self.assert_row_count(project, "microbatch_model", 3)
+
+
+class TestMicrbobatchModelsRunWithSameCurrentTime(BaseMicrobatchTest):
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_sql,
+            "second_microbatch_model.sql": microbatch_model_downstream_sql,
+        }
+
+    @mock.patch.dict(os.environ, {"DBT_EXPERIMENTAL_MICROBATCH": "True"})
+    def test_microbatch(self, project) -> None:
+        current_time = datetime.now(pytz.UTC)
+        run_dbt(["run", "--event-time-start", current_time.strftime("%Y-%m-%d")])
+
+        run_results = get_artifact(project.project_root, "target", "run_results.json")
+        microbatch_model_last_batch = run_results["results"][1]["batch_results"]["successful"][-1]
+        second_microbatch_model_last_batch = run_results["results"][2]["batch_results"][
+            "successful"
+        ][-1]
+
+        # they should have the same last batch because they are using the _same_ "current_time"
+        assert microbatch_model_last_batch == second_microbatch_model_last_batch


Code review comments:
diff_hunk: @@ -98,6 +101,7 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
     profile_name: str
     cli_vars: Dict[str, Any]
     dependencies: Optional[Mapping[str, "RuntimeConfig"]] = None
+    invocated_at: datetime = field(default_factory=lambda: datetime.now(pytz.UTC))
path: core/dbt/config/runtime.py
text: @user1:
Can we simplify this to `invoked_at`?

@author:
We can 🙂 I went back and for on this rename myself. Will get it changed.

diff_hunk: @@ -35,10 +36,13 @@ def __init__(
             event_time_start.replace(tzinfo=pytz.UTC) if event_time_start else None
         )
         self.event_time_end = event_time_end.replace(tzinfo=pytz.UTC) if event_time_end else None
+        self.batch_current_time = (
+            batch_current_time if batch_current_time is not None else datetime.now(pytz.UTC)
+        )
path: core/dbt/materializations/incremental/microbatch.py
text: @user1:
Nit: can be simplified to 

```suggestion
        self.batch_current_time = (
            batch_current_time or datetime.now(pytz.UTC)
        )
```

diff_hunk: @@ -35,10 +36,13 @@ def __init__(
             event_time_start.replace(tzinfo=pytz.UTC) if event_time_start else None
         )
         self.event_time_end = event_time_end.replace(tzinfo=pytz.UTC) if event_time_end else None
+        self.batch_current_time = (
+            batch_current_time if batch_current_time is not None else datetime.now(pytz.UTC)
+        )
 
     def build_end_time(self):
         """Defaults the end_time to the current time in UTC unless a non `None` event_time_end was provided"""
-        return self.event_time_end or datetime.now(tz=pytz.utc)
+        return self.event_time_end or self.batch_current_time
path: core/dbt/materializations/incremental/microbatch.py
text: @user1:
Under which situations will we return `self.batch_current_time` here?
Probably worth adding to the description.

@author:
From the function perspective, we return `self.batch_current_time` when `self.event_time_end` is `None`. From a system perspective this currently happens when the dbt runner is not invoked with an `--event-time-end <my-desired-end-value>`. I don't think the system context is relevant to this function though. If anything I think we should separately push that context to the instantiation of the class in `run.py` and drop this `or`-ing from the class itself.

I do have a planned refactor of this class for this week, this context shifting would be more appropriate as part of that.

diff_hunk: @@ -18,6 +18,7 @@ def __init__(
         is_incremental: bool,
         event_time_start: Optional[datetime],
         event_time_end: Optional[datetime],
+        batch_current_time: Optional[datetime] = None,
path: core/dbt/materializations/incremental/microbatch.py
text: @user1:
Can this be `batch_invoked_at` for consistency?

@author:
I'm down to rename the variable. However, I think `batch_invoked_at` would give the wrong impression. This datetime is not when the batch was actually invoked. The datetime will be before the batch is actually invoked, by how much is dependent on how long the project takes to run and when the microbatch model is being worked on within that.

@author:
How about `batch_default_end_time`?

@author:
or `default_end_time`? 🤔 Saying `batch` feels kinda redundant