Repository: dbt-labs/dbt-core
Pull request: Add required 'begin' config support for microbatch models
Resolves https://github.com/dbt-labs/dbt-core/issues/10701

<!---
  Include the number of the issue addressed by this PR above, if applicable.
  PRs for code changes without an associated issue *will not be merged*.
  See CONTRIBUTING.md for more information.

  Add the `user docs` label to this PR if it will need docs changes.  An 
  issue will get opened in docs.getdbt.com upon successful merge of this PR.
-->

### Problem

<!---
  Describe the problem this PR is solving. What is the application state
  before this PR is merged?
-->

We need to know what the "beginning of time" is for a microbatch model, to know when to start for these two cases:
* The first time this model is run
* If you run in “full-refresh-mode”

This is distinct from the CLI parameter --event-time-start which can also be used when running in "incremental" mode.

### Solution

<!---
  Describe the way this PR solves the above problem. Add as much detail as you
  can to help reviewers understand your changes. Include any alternatives and
  tradeoffs you considered.
-->
* continue to return early if --event-time-start is provided
* For non-incremental runs (or if there is no checkpoint to lookback from), return the `begin` config! (raise exception if its not available, and further parse-time validation will be introduced as part of https://github.com/dbt-labs/dbt-core/pull/10752
* update schemas.getdbt.com v12 manifest
* update tests to provide `begin` config.

### 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:
Require defining a `begin` config for microbatch incremental models
We need to know what the "beginning of time" is for a microbatch model, to know when to start for these two cases:
- The first time this model is run
- If you run in “full-refresh-mode”

This is distinct from the CLI parameter `--event-time-start` which can also be used when running in "incremental" mode.

Hints from issue discussion:

Commit message: fix functional tests

Patch to review:
diff --git a/.changes/unreleased/Features-20240923-155903.yaml b/.changes/unreleased/Features-20240923-155903.yaml
new file mode 100644
index 00000000000..b05e81b94d2
--- /dev/null
+++ b/.changes/unreleased/Features-20240923-155903.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Support required 'begin' config for microbatch models
+time: 2024-09-23T15:59:03.924079+01:00
+custom:
+  Author: michelleark
+  Issue: "10701"
diff --git a/core/dbt/artifacts/resources/v1/config.py b/core/dbt/artifacts/resources/v1/config.py
index e6cd26ec823..5a7977c0958 100644
--- a/core/dbt/artifacts/resources/v1/config.py
+++ b/core/dbt/artifacts/resources/v1/config.py
@@ -82,6 +82,7 @@ class NodeConfig(NodeAndTestConfig):
     incremental_strategy: Optional[str] = None
     batch_size: Any = None
     lookback: Any = 0
+    begin: Any = None
     persist_docs: Dict[str, Any] = field(default_factory=dict)
     post_hook: List[Hook] = field(
         default_factory=list,
diff --git a/core/dbt/materializations/incremental/microbatch.py b/core/dbt/materializations/incremental/microbatch.py
index 5bd46eae5e9..524cb8ab009 100644
--- a/core/dbt/materializations/incremental/microbatch.py
+++ b/core/dbt/materializations/incremental/microbatch.py
@@ -46,41 +46,39 @@ def build_start_time(self, checkpoint: Optional[datetime]):
         to build a start time. This is because we build the start time relative to the checkpoint
         via the batchsize and offset, and we cannot offset a checkpoint if there is no checkpoint.
         """
+        assert isinstance(self.model.config, NodeConfig)
+        batch_size = self.model.config.batch_size
 
+        # Use event_time_start if it is provided.
         if self.event_time_start:
-            return MicrobatchBuilder.truncate_timestamp(
-                self.event_time_start, self.model.config.batch_size
-            )
+            return MicrobatchBuilder.truncate_timestamp(self.event_time_start, batch_size)
 
+        # First run, use model's configured 'begin' as start.
         if not self.is_incremental or checkpoint is None:
-            # TODO: return new model-level configuration or raise error
-            return None
+            if not self.model.config.begin:
+                raise DbtRuntimeError(
+                    f"Microbatch model '{self.model.name}' requires a 'begin' configuration."
+                )
 
-        assert isinstance(self.model.config, NodeConfig)
-        batch_size = self.model.config.batch_size
+            return MicrobatchBuilder.truncate_timestamp(self.model.config.begin, batch_size)
 
         lookback = self.model.config.lookback
         start = MicrobatchBuilder.offset_timestamp(checkpoint, batch_size, -1 * lookback)
 
         return start
 
-    def build_batches(
-        self, start: Optional[datetime], end: datetime
-    ) -> List[Tuple[Optional[datetime], datetime]]:
+    def build_batches(self, start: datetime, end: datetime) -> List[Tuple[datetime, datetime]]:
         """
         Given a start and end datetime, builds a list of batches where each batch is
         the size of the model's batch_size.
         """
-        if start is None:
-            return [(start, end)]
-
         batch_size = self.model.config.batch_size
         curr_batch_start: datetime = start
         curr_batch_end: datetime = MicrobatchBuilder.offset_timestamp(
             curr_batch_start, batch_size, 1
         )
 
-        batches: List[Tuple[Optional[datetime], datetime]] = [(curr_batch_start, curr_batch_end)]
+        batches: List[Tuple[datetime, datetime]] = [(curr_batch_start, curr_batch_end)]
         while curr_batch_end <= end:
             curr_batch_start = curr_batch_end
             curr_batch_end = MicrobatchBuilder.offset_timestamp(curr_batch_start, batch_size, 1)
diff --git a/tests/functional/artifacts/expected_manifest.py b/tests/functional/artifacts/expected_manifest.py
index 749a5b77619..606a3e8f5d7 100644
--- a/tests/functional/artifacts/expected_manifest.py
+++ b/tests/functional/artifacts/expected_manifest.py
@@ -42,6 +42,7 @@ def get_rendered_model_config(**updates):
         "event_time": None,
         "lookback": 0,
         "batch_size": None,
+        "begin": None,
     }
     result.update(updates)
     return result
@@ -80,6 +81,7 @@ def get_rendered_seed_config(**updates):
         "event_time": None,
         "lookback": 0,
         "batch_size": None,
+        "begin": None,
     }
     result.update(updates)
     return result
@@ -129,6 +131,7 @@ def get_rendered_snapshot_config(**updates):
         "event_time": None,
         "lookback": 0,
         "batch_size": None,
+        "begin": None,
     }
     result.update(updates)
     return result
diff --git a/tests/functional/list/test_list.py b/tests/functional/list/test_list.py
index 84b1e382e89..6894ec96bb4 100644
--- a/tests/functional/list/test_list.py
+++ b/tests/functional/list/test_list.py
@@ -88,6 +88,7 @@ def expect_snapshot_output(self, happy_path_project):  # noqa: F811
                     "event_time": None,
                     "lookback": 0,
                     "batch_size": None,
+                    "begin": None,
                 },
                 "unique_id": "snapshot.test.my_snapshot",
                 "original_file_path": normalize("snapshots/snapshot.sql"),
@@ -133,6 +134,7 @@ def expect_analyses_output(self):
                     "event_time": None,
                     "lookback": 0,
                     "batch_size": None,
+                    "begin": None,
                 },
                 "unique_id": "analysis.test.a",
                 "original_file_path": normalize("analyses/a.sql"),
@@ -197,6 +199,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/ephemeral.sql"),
                     "unique_id": "model.test.ephemeral",
@@ -238,6 +241,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/incremental.sql"),
                     "unique_id": "model.test.incremental",
@@ -279,6 +283,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/sub/inner.sql"),
                     "unique_id": "model.test.inner",
@@ -320,6 +325,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/metricflow_time_spine.sql"),
                     "unique_id": "model.test.metricflow_time_spine",
@@ -361,6 +367,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/metricflow_time_spine_second.sql"),
                     "unique_id": "model.test.metricflow_time_spine_second",
@@ -402,6 +409,7 @@ def expect_model_output(self):
                         "event_time": None,
                         "lookback": 0,
                         "batch_size": None,
+                        "begin": None,
                     },
                     "original_file_path": normalize("models/outer.sql"),
                     "unique_id": "model.test.outer",
@@ -524,6 +532,7 @@ def expect_seed_output(self):
                     "event_time": None,
                     "lookback": 0,
                     "batch_size": None,
+                    "begin": None,
                 },
                 "depends_on": {"macros": []},
                 "unique_id": "seed.test.seed",
diff --git a/tests/functional/microbatch/test_microbatch.py b/tests/functional/microbatch/test_microbatch.py
index cf8e018727f..aea76fafb8a 100644
--- a/tests/functional/microbatch/test_microbatch.py
+++ b/tests/functional/microbatch/test_microbatch.py
@@ -32,7 +32,7 @@
 """
 
 microbatch_model_sql = """
-{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='day') }}
+{{ 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('input_model') }}
 """
 
@@ -67,7 +67,7 @@
 """
 
 microbatch_model_calling_source_sql = """
-{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='day') }}
+{{ 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 {{ source('seed_sources', 'raw_source') }}
 """
 
@@ -122,10 +122,12 @@ def test_use_custom_microbatch_strategy_env_var_true_invalid_incremental_strateg
             type(project.adapter), "valid_incremental_strategies", lambda _: ["microbatch"]
         ):
             # Initial run
-            run_dbt(["run"])
+            with patch_microbatch_end_time("2020-01-03 13:57:00"):
+                run_dbt(["run"])
 
             # Incremental run uses custom strategy
-            _, logs = run_dbt_and_capture(["run"])
+            with patch_microbatch_end_time("2020-01-03 13:57:00"):
+                _, logs = run_dbt_and_capture(["run"])
             assert "custom microbatch strategy" in logs
 
 
@@ -140,10 +142,12 @@ def test_use_custom_microbatch_strategy_env_var_true_invalid_incremental_strateg
             type(project.adapter), "valid_incremental_strategies", lambda _: []
         ):
             # Initial run
-            run_dbt(["run"])
+            with patch_microbatch_end_time("2020-01-03 13:57:00"):
+                run_dbt(["run"])
 
             # Incremental run fails
-            _, logs = run_dbt_and_capture(["run"], expect_pass=False)
+            with patch_microbatch_end_time("2020-01-03 13:57:00"):
+                _, logs = run_dbt_and_capture(["run"], expect_pass=False)
             assert "'microbatch' is not valid" in logs
 
 
diff --git a/tests/unit/materializations/incremental/test_microbatch.py b/tests/unit/materializations/incremental/test_microbatch.py
index 68521a84e1e..5b4a959ca41 100644
--- a/tests/unit/materializations/incremental/test_microbatch.py
+++ b/tests/unit/materializations/incremental/test_microbatch.py
@@ -9,6 +9,8 @@
 from dbt.artifacts.resources.types import BatchSize
 from dbt.materializations.incremental.microbatch import MicrobatchBuilder
 
+MODEL_CONFIG_BEGIN = datetime(2024, 1, 1, 0, 0, 0, 0, pytz.UTC)
+
 
 class TestMicrobatchBuilder:
     @pytest.fixture(scope="class")
@@ -17,6 +19,7 @@ def microbatch_model(self):
         model.config = mock.MagicMock(NodeConfig)
         model.config.materialized = "incremental"
         model.config.incremental_strategy = "microbatch"
+        model.config.begin = MODEL_CONFIG_BEGIN
 
         return model
 
@@ -67,7 +70,8 @@ def test_build_end_time(
                 datetime(2024, 9, 5, 8, 56, 0, 0, pytz.UTC),
                 BatchSize.day,
                 0,
-                None,
+                # is_incremental: False => model.config.begin
+                MODEL_CONFIG_BEGIN,
             ),
             # BatchSize.year
             (
@@ -93,8 +97,8 @@ def test_build_end_time(
                 datetime(2024, 9, 5, 8, 56, 0, 0, pytz.UTC),
                 BatchSize.year,
                 0,
-                # is_incremental=False + no start_time -> None
-                None,
+                # is_incremental=False + no start_time -> model.config.begin
+                MODEL_CONFIG_BEGIN,
             ),
             (
                 True,
@@ -136,8 +140,8 @@ def test_build_end_time(
                 datetime(2024, 9, 5, 8, 56, 0, 0, pytz.UTC),
                 BatchSize.month,
                 0,
-                # is_incremental=False + no start_time -> None
-                None,
+                # is_incremental=False + no start_time -> model.config.begin
+                MODEL_CONFIG_BEGIN,
             ),
             (
                 True,
@@ -179,8 +183,8 @@ def test_build_end_time(
                 datetime(2024, 9, 5, 8, 56, 0, 0, pytz.UTC),
                 BatchSize.day,
                 0,
-                # is_incremental=False + no start_time -> None
-                None,
+                # is_incremental=False + no start_time -> model.config.begin
+                MODEL_CONFIG_BEGIN,
             ),
             (
                 True,
@@ -222,8 +226,8 @@ def test_build_end_time(
                 datetime(2024, 9, 5, 8, 56, 0, 0, pytz.UTC),
                 BatchSize.hour,
                 0,
-                # is_incremental=False + no start_time -> None
-                None,
+                # is_incremental=False + no start_time -> model.config.begin
+                MODEL_CONFIG_BEGIN,
             ),
             (
                 True,


Code review comments:
diff_hunk: @@ -46,41 +46,39 @@ def build_start_time(self, checkpoint: Optional[datetime]):
         to build a start time. This is because we build the start time relative to the checkpoint
path: core/dbt/materializations/incremental/microbatch.py
text: @user1:
We should update this doc string for the change functionality. `None` is never returned now I believe.