raise MicrobatchModelNoEventTimeInputs warning when no microbatch inp…

…ut has event_time config

core/dbt/events/core_types.proto

@@ -893,6 +893,17 @@ message FreshnessConfigProblemMsg {
 }
 
 
+// I074
+message MicrobatchModelNoEventTimeInputs {
+    string model_name = 1;
+}
+
+message MicrobatchModelNoEventTimeInputsMsg {
+    CoreEventInfo info = 1;
+    MicrobatchModelNoEventTimeInputs data = 2;
+}
+
+
 // M - Deps generation
 
 

core/dbt/events/types.py

@@ -924,6 +924,19 @@ def message(self) -> str:
         return self.msg
 
 
+class MicrobatchModelNoEventTimeInputs(WarnLevel):
+    def code(self) -> str:
+        return "I074"
+
+    def message(self) -> str:
+        msg = (
+            f"The microbatch model '{self.model_name}' has no 'ref' or 'source' input with an 'event_time' configuration. "
+            "This can result in unexpected duplicate records in the resulting microbatch model."
+        )
+
+        return warning_tag(msg)
+
+
 # =======================================================
 # M - Deps generation
 # =======================================================

core/dbt/parser/manifest.py

@@ -66,6 +66,7 @@
     DeprecatedModel,
     DeprecatedReference,
     InvalidDisabledTargetInTestNode,
+    MicrobatchModelNoEventTimeInputs,
     NodeNotFoundOrDisabled,
     ParsedFileLoadFailed,
     ParsePerfInfoPath,
@@ -1442,13 +1443,19 @@ def check_valid_microbatch_config(self):
                         )
 
                     # Validate upstream node event_time (if configured)
+                    has_input_with_event_time_config = False
                     for input_unique_id in node.depends_on.nodes:
                         input_node = self.manifest.expect(unique_id=input_unique_id)
                         input_event_time = input_node.config.event_time
-                        if input_event_time and not isinstance(input_event_time, str):
-                            raise dbt.exceptions.ParsingError(
-                                f"Microbatch model '{node.name}' depends on an input node '{input_node.name}' with an 'event_time' config of invalid (non-string) type: {type(input_event_time)}."
-                            )
+                        if input_event_time:
+                            if not isinstance(input_event_time, str):
+                                raise dbt.exceptions.ParsingError(
+                                    f"Microbatch model '{node.name}' depends on an input node '{input_node.name}' with an 'event_time' config of invalid (non-string) type: {type(input_event_time)}."
+                                )
+                            has_input_with_event_time_config = True
+
+                    if not has_input_with_event_time_config:
+                        fire_event(MicrobatchModelNoEventTimeInputs(model_name=node.name))
 
     def write_perf_info(self, target_path: str):
         path = os.path.join(target_path, PERF_INFO_FILE_NAME)

tests/functional/microbatch/test_microbatch.py

@@ -5,7 +5,7 @@
 import pytest
 import pytz
 
-from dbt.events.types import LogModelResult
+from dbt.events.types import LogModelResult, MicrobatchModelNoEventTimeInputs
 from dbt.tests.util import (
     get_artifact,
     patch_microbatch_end_time,
@@ -47,7 +47,6 @@
 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)) }}
 select * from {{ ref('input_model').render() }}
@@ -290,9 +289,11 @@ def test_run_with_event_time(self, project):
         run_dbt(["seed"])
 
         # initial run -- backfills all data
+        catcher = EventCatcher(event_to_catch=MicrobatchModelNoEventTimeInputs)
         with patch_microbatch_end_time("2020-01-03 13:57:00"):
-            run_dbt(["run"])
+            run_dbt(["run"], callbacks=[catcher.catch])
         self.assert_row_count(project, "microbatch_model", 3)
+        assert len(catcher.caught_events) == 0
 
         # our partition grain is "day" so running the same day without new data should produce the same results
         with patch_microbatch_end_time("2020-01-03 14:57:00"):
@@ -334,10 +335,13 @@ def models(self):
 
     @mock.patch.dict(os.environ, {"DBT_EXPERIMENTAL_MICROBATCH": "True"})
     def test_run_with_event_time(self, project):
+        catcher = EventCatcher(event_to_catch=MicrobatchModelNoEventTimeInputs)
+
         # initial run -- backfills all data
         with patch_microbatch_end_time("2020-01-03 13:57:00"):
-            run_dbt(["run"])
+            run_dbt(["run"], callbacks=[catcher.catch])
         self.assert_row_count(project, "microbatch_model", 3)
+        assert len(catcher.caught_events) == 1
 
         # our partition grain is "day" so running the same day without new data should produce the same results
         with patch_microbatch_end_time("2020-01-03 14:57:00"):