Fix microbatch file naming for compiled code (#11220)

* Functional test for hourly microbatch model

* Use today's date for functional test for hourly microbatch model

* Use today's date for functional test for hourly microbatch model

* Restore to original

* Only use alphanumeric characters within batch ids

* Add tests for batch_id and change expected output for format_batch_start

* Handle missing batch_start

* Revert "Handle missing batch_start"

This reverts commit 65a1db0. Reverting this because
`batch_start` for `format_batch_start` cannot be `None` and `start_time` for `batch_id`
cannot be `None`.

* Improve BatchSize specific values for `format_batch_start` and `batch_id` methods

---------

Co-authored-by: Quigley Malcolm <[email protected]>

Author: dbeatty10

core/dbt/context/providers.py

@@ -1050,9 +1050,10 @@ def write(self, payload: str) -> str:
         if (
             isinstance(self.model, ModelNode)
             and self.model.config.get("incremental_strategy") == "microbatch"
+            and self.model.batch is not None
         ):
             split_suffix = MicrobatchBuilder.format_batch_start(
-                self.model.config.get("__dbt_internal_microbatch_event_time_start"),
+                self.model.batch.event_time_start,
                 self.model.config.batch_size,
             )
 

core/dbt/materializations/incremental/microbatch.py

@@ -199,9 +199,21 @@ def batch_id(start_time: datetime, batch_size: BatchSize) -> str:
 
     @staticmethod
     def format_batch_start(batch_start: datetime, batch_size: BatchSize) -> str:
-        return str(
-            batch_start.date() if (batch_start and batch_size != BatchSize.hour) else batch_start
-        )
+        """Format the passed in datetime based on the batch_size.
+
+        2024-09-17 16:06:00 + Batchsize.hour  -> 2024-09-17T16
+        2024-09-17 16:06:00 + Batchsize.day   -> 2024-09-17
+        2024-09-17 16:06:00 + Batchsize.month -> 2024-09
+        2024-09-17 16:06:00 + Batchsize.year  -> 2024
+        """
+        if batch_size == BatchSize.year:
+            return batch_start.strftime("%Y")
+        elif batch_size == BatchSize.month:
+            return batch_start.strftime("%Y-%m")
+        elif batch_size == BatchSize.day:
+            return batch_start.strftime("%Y-%m-%d")
+        else:  # batch_size == BatchSize.hour
+            return batch_start.strftime("%Y-%m-%dT%H")
 
     @staticmethod
     def ceiling_timestamp(timestamp: datetime, batch_size: BatchSize) -> datetime:

tests/functional/microbatch/test_microbatch.py

@@ -57,6 +57,21 @@
 select * from {{ ref('input_model') }}
 """
 
+microbatch_model_hour_sql = """
+{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='hour', begin=modules.datetime.datetime(2020, 1, 1, 0, 0, 0)) }}
+select * from {{ ref('input_model') }}
+"""
+
+microbatch_model_month_sql = """
+{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='month', begin=modules.datetime.datetime(2020, 1, 1, 0, 0, 0)) }}
+select * from {{ ref('input_model') }}
+"""
+
+microbatch_model_year_sql = """
+{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='event_time', batch_size='year', begin=modules.datetime.datetime(2020, 1, 1, 0, 0, 0)) }}
+select * from {{ ref('input_model') }}
+"""
+
 microbatch_model_with_pre_and_post_sql = """
 {{ config(
         materialized='incremental',
@@ -841,6 +856,111 @@ def test_run_with_event_time(self, project):
         )
 
 
+class TestMicrobatchCompiledRunPathsHourly(BaseMicrobatchTest):
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_hour_sql,
+        }
+
+    def test_run_with_event_time(self, project):
+        # run all partitions from start - 2 expected rows in output, one failed
+        with patch_microbatch_end_time("2020-01-03 13:57:00"):
+            run_dbt(["run"])
+
+        # Compiled paths - batch compilations
+        assert read_file(
+            project.project_root,
+            "target",
+            "compiled",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020-01-03T13.sql",
+        )
+        assert read_file(
+            project.project_root,
+            "target",
+            "run",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020-01-03T13.sql",
+        )
+
+
+class TestMicrobatchCompiledRunPathsMonthly(BaseMicrobatchTest):
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_month_sql,
+        }
+
+    def test_run_with_event_time(self, project):
+        # run all partitions from start - 2 expected rows in output, one failed
+        with patch_microbatch_end_time("2020-01-03 13:57:00"):
+            run_dbt(["run"])
+
+        # Compiled paths - batch compilations
+        assert read_file(
+            project.project_root,
+            "target",
+            "compiled",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020-01.sql",
+        )
+        assert read_file(
+            project.project_root,
+            "target",
+            "run",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020-01.sql",
+        )
+
+
+class TestMicrobatchCompiledRunPathsYearly(BaseMicrobatchTest):
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": input_model_sql,
+            "microbatch_model.sql": microbatch_model_year_sql,
+        }
+
+    def test_run_with_event_time(self, project):
+        # run all partitions from start - 2 expected rows in output, one failed
+        with patch_microbatch_end_time("2020-01-03 13:57:00"):
+            run_dbt(["run"])
+
+        # Compiled paths - batch compilations
+        assert read_file(
+            project.project_root,
+            "target",
+            "compiled",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020.sql",
+        )
+        assert read_file(
+            project.project_root,
+            "target",
+            "run",
+            "test",
+            "models",
+            "microbatch_model",
+            "microbatch_model_2020.sql",
+        )
+
+
 class TestMicrobatchFullRefreshConfigFalse(BaseMicrobatchTest):
     @pytest.fixture(scope="class")
     def models(self):

tests/unit/materializations/incremental/test_microbatch.py

@@ -602,16 +602,32 @@ def test_offset_timestamp(self, timestamp, batch_size, offset, expected_timestam
     def test_truncate_timestamp(self, timestamp, batch_size, expected_timestamp):
         assert MicrobatchBuilder.truncate_timestamp(timestamp, batch_size) == expected_timestamp
 
+    @pytest.mark.parametrize(
+        "batch_size,start_time,expected_formatted_start_time",
+        [
+            (BatchSize.year, datetime(2020, 1, 1, 1), "2020"),
+            (BatchSize.month, datetime(2020, 1, 1, 1), "202001"),
+            (BatchSize.day, datetime(2020, 1, 1, 1), "20200101"),
+            (BatchSize.hour, datetime(2020, 1, 1, 1), "20200101T01"),
+        ],
+    )
+    def test_batch_id(
+        self, batch_size: BatchSize, start_time: datetime, expected_formatted_start_time: str
+    ) -> None:
+        assert MicrobatchBuilder.batch_id(start_time, batch_size) == expected_formatted_start_time
+
     @pytest.mark.parametrize(
         "batch_size,batch_start,expected_formatted_batch_start",
         [
-            (BatchSize.year, datetime(2020, 1, 1, 1), "2020-01-01"),
-            (BatchSize.month, datetime(2020, 1, 1, 1), "2020-01-01"),
+            (BatchSize.year, datetime(2020, 1, 1, 1), "2020"),
+            (BatchSize.month, datetime(2020, 1, 1, 1), "2020-01"),
             (BatchSize.day, datetime(2020, 1, 1, 1), "2020-01-01"),
-            (BatchSize.hour, datetime(2020, 1, 1, 1), "2020-01-01 01:00:00"),
+            (BatchSize.hour, datetime(2020, 1, 1, 1), "2020-01-01T01"),
         ],
     )
-    def test_format_batch_start(self, batch_size, batch_start, expected_formatted_batch_start):
+    def test_format_batch_start(
+        self, batch_size: BatchSize, batch_start: datetime, expected_formatted_batch_start: str
+    ) -> None:
         assert (
             MicrobatchBuilder.format_batch_start(batch_start, batch_size)
             == expected_formatted_batch_start