Fix: Propagation of the forward_only flag from dbt models when it's set to false (#4121)

Author: izeigerman

sqlmesh/dbt/model.py

@@ -238,8 +238,10 @@ def model_kind(self, context: DbtContext) -> ModelKind:
             incremental_kind_kwargs["on_destructive_change"] = on_destructive_change
 
         for field in ("forward_only", "auto_restatement_cron"):
-            field_val = getattr(self, field, None) or self.meta.get(field, None)
-            if field_val:
+            field_val = getattr(self, field, None)
+            if field_val is None:
+                field_val = self.meta.get(field, None)
+            if field_val is not None:
                 incremental_kind_kwargs[field] = field_val
 
         if materialization == Materialization.TABLE:
@@ -249,8 +251,10 @@ def model_kind(self, context: DbtContext) -> ModelKind:
         if materialization == Materialization.INCREMENTAL:
             incremental_by_kind_kwargs: t.Dict[str, t.Any] = {"dialect": self.dialect(context)}
             for field in ("batch_size", "batch_concurrency", "lookback"):
-                field_val = getattr(self, field, None) or self.meta.get(field, None)
-                if field_val:
+                field_val = getattr(self, field, None)
+                if field_val is None:
+                    field_val = self.meta.get(field, None)
+                if field_val is not None:
                     incremental_by_kind_kwargs[field] = field_val
 
             if self.time_column:

tests/core/test_macros.py

@@ -695,7 +695,7 @@ def test_macro_parameter_resolution(macro_evaluator):
     # with the former one. This ensures we capture both.
     assert str(e.value.__cause__) in (
         "'pos_only' parameter is positional only, but was passed as a keyword",
-        "missing a required argument: 'a1'",
+        "missing a required positional-only argument: 'pos_only'",
     )
 
     with pytest.raises(MacroEvalError) as e:

tests/dbt/test_transformation.py

@@ -286,6 +286,12 @@ def test_model_kind():
         context
     ) == IncrementalUnmanagedKind(insert_overwrite=True, disable_restatement=False)
 
+    assert ModelConfig(materialized=Materialization.INCREMENTAL, forward_only=False).model_kind(
+        context
+    ) == IncrementalUnmanagedKind(
+        insert_overwrite=True, disable_restatement=False, forward_only=False
+    )
+
     assert ModelConfig(
         materialized=Materialization.INCREMENTAL, incremental_strategy="append"
     ).model_kind(context) == IncrementalUnmanagedKind(disable_restatement=False)