Fix: Use the interval_unit instead of cron when normalizing intervals during evaluation in Airflow (#1612)

Author: izeigerman

sqlmesh/schedulers/airflow/operators/hwm_sensor.py

@@ -65,8 +65,8 @@ def _compute_target_high_water_mark(
         self, dag_run: DagRun, target_snapshot: Snapshot
     ) -> datetime:
         target_date = to_datetime(dag_run.data_interval_end)
-        target_prev = to_datetime(target_snapshot.node.cron_floor(target_date))
-        this_prev = to_datetime(self.this_snapshot.node.cron_floor(target_date))
+        target_prev = to_datetime(target_snapshot.node.interval_unit.cron_floor(target_date))
+        this_prev = to_datetime(self.this_snapshot.node.interval_unit.cron_floor(target_date))
         return min(target_prev, this_prev)
 
 

sqlmesh/schedulers/airflow/operators/targets.py

@@ -1,6 +1,5 @@
 import abc
 import typing as t
-from datetime import datetime
 
 from airflow.exceptions import AirflowSkipException
 from airflow.utils.context import Context
@@ -154,14 +153,16 @@ def _get_start(self, context: Context) -> TimeLike:
         if self.start:
             return self.start
 
-        start = t.cast(datetime, context["dag_run"].data_interval_start)
+        start = self.snapshot.node.interval_unit.cron_floor(context["dag_run"].data_interval_start)
         if not self.snapshot.is_model:
             return start
 
         return self.snapshot.model.lookback_start(start)
 
     def _get_end(self, context: Context) -> TimeLike:
-        return self.end or context["dag_run"].data_interval_end
+        return self.end or self.snapshot.node.interval_unit.cron_floor(
+            context["dag_run"].data_interval_end
+        )
 
     def _get_execution_time(self, context: Context) -> TimeLike:
         return self.execution_time or context["dag_run"].logical_date