Feat dbt incremental allow partials (#3709)

Author: izeigerman

docs/integrations/dbt.md

@@ -201,6 +201,8 @@ The [`lookback` parameter](../concepts/models/overview.md#lookback) is used to c
 
 **Note:** By default, all incremental dbt models are configured to be [forward-only](../concepts/plans.md#forward-only-plans). However, you can change this behavior by setting the `forward_only: false` setting either in the configuration of an individual model or globally for all models in the `dbt_project.yaml` file. The [forward-only](../concepts/plans.md#forward-only-plans) mode aligns more closely with the typical operation of dbt and therefore better meets user's expectations.
 
+Similarly, the [allow_partials](../concepts/models/overview.md#allow_partials) parameter is set to `true` by default for incremental dbt models unless the time column is specified, or the `allow_partials` parameter is explicitly set to `false` in the model configuration.
+
 #### on_schema_change
 
 SQLMesh automatically detects destructive schema changes to [forward-only incremental models](../guides/incremental_time.md#forward-only-models) and to all incremental models in [forward-only plans](../concepts/plans.md#destructive-changes).

sqlmesh/dbt/common.py

@@ -149,8 +149,10 @@ def sqlmesh_config_kwargs(self) -> t.Dict[str, t.Any]:
         """
         kwargs = {}
         for field in self.sqlmesh_config_fields:
-            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:
                 kwargs[field] = field_val
         return kwargs
 

sqlmesh/dbt/model.py

@@ -554,17 +554,26 @@ def to_sqlmesh(
             if physical_properties:
                 model_kwargs["physical_properties"] = physical_properties
 
+        kind = self.model_kind(context)
+        allow_partials = model_kwargs.pop("allow_partials", None)
+        if allow_partials is None and (
+            kind.is_incremental_unmanaged or kind.is_incremental_by_unique_key
+        ):
+            # Set allow_partials to True for dbt incremental models to preserve the original semantics.
+            allow_partials = True
+
         model = create_sql_model(
             self.canonical_name(context),
             query,
             dialect=model_dialect,
-            kind=self.model_kind(context),
+            kind=kind,
             start=self.start,
             audit_definitions=audit_definitions,
             # This ensures that we bypass query rendering that would otherwise be required to extract additional
             # dependencies from the model's SQL.
             # Note: any table dependencies that are not referenced using the `ref` macro will not be included.
             extract_dependencies_from_query=False,
+            allow_partials=allow_partials,
             **optional_kwargs,
             **model_kwargs,
         )

tests/dbt/test_transformation.py

@@ -1450,3 +1450,36 @@ def test_refs_in_jinja_globals(sushi_test_project: Project, mocker: MockerFixtur
         "waiter_revenue_by_day",
         "sushi.waiter_revenue_by_day",
     }
+
+
+def test_dbt_incremental_allow_partials_by_default():
+    context = DbtContext()
+    context._target = SnowflakeConfig(
+        name="target",
+        schema="test",
+        database="test",
+        account="account",
+        user="user",
+        password="password",
+    )
+
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        sql="SELECT * FROM baz",
+        materialized=Materialization.TABLE.value,
+    )
+    assert model.allow_partials is None
+    assert not model.to_sqlmesh(context).allow_partials
+
+    model.materialized = Materialization.INCREMENTAL.value
+    assert model.allow_partials is None
+    assert model.to_sqlmesh(context).allow_partials
+
+    model.allow_partials = True
+    assert model.to_sqlmesh(context).allow_partials
+
+    model.allow_partials = False
+    assert not model.to_sqlmesh(context).allow_partials