Set an invocated_at on the RuntimeConfig and plumb to `Microbatch…

…Builder`

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()

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.

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)