[ADAP-631] Convert Target_lag to str type to add Downstream option

.changes/unreleased/Fixes-20230810-154613.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: change target_lag type to allow for downstream as a option
+time: 2023-08-10T15:46:13.896057-05:00
+custom:
+  Author: McKnight-42
+  Issue: "734"

dbt/adapters/snowflake/relation_configs/__init__.py

@@ -2,14 +2,10 @@
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
     SnowflakeDynamicTableWarehouseConfigChange,
+    SnowflakeDynamicTableTargetLagConfigChange,
 )
 from dbt.adapters.snowflake.relation_configs.policies import (
     SnowflakeIncludePolicy,
     SnowflakeQuotePolicy,
     SnowflakeRelationType,
 )
-from dbt.adapters.snowflake.relation_configs.target_lag import (
-    SnowflakeDynamicTableTargetLagConfig,
-    SnowflakeDynamicTableTargetLagConfigChange,
-    SnowflakeDynamicTableTargetLagPeriod,
-)

dbt/adapters/snowflake/relation_configs/dynamic_table.py

@@ -7,10 +7,6 @@
 from dbt.contracts.relation import ComponentName
 
 from dbt.adapters.snowflake.relation_configs.base import SnowflakeRelationConfigBase
-from dbt.adapters.snowflake.relation_configs.target_lag import (
-    SnowflakeDynamicTableTargetLagConfig,
-    SnowflakeDynamicTableTargetLagConfigChange,
-)
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
@@ -32,7 +28,7 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     schema_name: str
     database_name: str
     query: str
-    target_lag: SnowflakeDynamicTableTargetLagConfig
+    target_lag: str
     snowflake_warehouse: str
 
     @classmethod
@@ -44,14 +40,10 @@ def from_dict(cls, config_dict) -> "SnowflakeDynamicTableConfig":
                 ComponentName.Database, config_dict.get("database_name")
             ),
             "query": config_dict.get("query"),
+            "target_lag": config_dict.get("target_lag"),
             "snowflake_warehouse": config_dict.get("snowflake_warehouse"),
         }
 
-        if target_lag := config_dict.get("target_lag"):
-            kwargs_dict.update(
-                {"target_lag": SnowflakeDynamicTableTargetLagConfig.from_dict(target_lag)}
-            )
-
         dynamic_table: "SnowflakeDynamicTableConfig" = super().from_dict(kwargs_dict)  # type: ignore
         return dynamic_table
 
@@ -62,14 +54,10 @@ def parse_model_node(cls, model_node: ModelNode) -> dict:
             "schema_name": model_node.schema,
             "database_name": model_node.database,
             "query": model_node.compiled_code,
+            "target_lag": model_node.config.extra.get("target_lag"),
             "snowflake_warehouse": model_node.config.extra.get("snowflake_warehouse"),
         }
 
-        if model_node.config.extra.get("target_lag"):
-            config_dict.update(
-                {"target_lag": SnowflakeDynamicTableTargetLagConfig.parse_model_node(model_node)}
-            )
-
         return config_dict
 
     @classmethod
@@ -81,21 +69,22 @@ def parse_relation_results(cls, relation_results: RelationResults) -> dict:
             "schema_name": dynamic_table.get("schema_name"),
             "database_name": dynamic_table.get("database_name"),
             "query": dynamic_table.get("text"),
+            "target_lag": dynamic_table.get("target_lag"),
             "snowflake_warehouse": dynamic_table.get("warehouse"),
         }
 
-        if dynamic_table.get("target_lag"):
-            config_dict.update(
-                {
-                    "target_lag": SnowflakeDynamicTableTargetLagConfig.parse_relation_results(
-                        dynamic_table
-                    )
-                }
-            )
-
         return config_dict
 
 
+@dataclass(frozen=True, eq=True, unsafe_hash=True)
+class SnowflakeDynamicTableTargetLagConfigChange(RelationConfigChange):
+    context: Optional[str] = None
+
+    @property
+    def requires_full_refresh(self) -> bool:
+        return False
+
+
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
 class SnowflakeDynamicTableWarehouseConfigChange(RelationConfigChange):
     context: Optional[str] = None