ADAP-1254: Add row access policy and table_tag config for snowflake (#901)

Co-authored-by: Colin Rogers <[email protected]>
Co-authored-by: Colin <[email protected]>
Co-authored-by: Mike Alfare <[email protected]>

Author: 3 people

dbt-snowflake/.changes/unreleased/Features-20250312-151053.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add row_access_policy and table_tag config for Snowflake models
+time: 2025-03-12T15:10:53.501206+01:00
+custom:
+  Author: b-per
+  Issue: "697"

dbt-snowflake/src/dbt/adapters/snowflake/impl.py

@@ -52,6 +52,8 @@ class SnowflakeConfig(AdapterConfig):
     tmp_relation_type: Optional[str] = None
     merge_update_columns: Optional[str] = None
     target_lag: Optional[str] = None
+    row_access_policy: Optional[str] = None
+    table_tag: Optional[str] = None
 
     # extended formats
     table_format: Optional[str] = None

dbt-snowflake/src/dbt/adapters/snowflake/relation_configs/dynamic_table.py

@@ -9,7 +9,6 @@
 
 from dbt.adapters.snowflake.relation_configs.base import SnowflakeRelationConfigBase
 
-
 if TYPE_CHECKING:
     import agate
 
@@ -58,6 +57,8 @@ class SnowflakeDynamicTableConfig(SnowflakeRelationConfigBase):
     snowflake_warehouse: str
     refresh_mode: Optional[RefreshMode] = RefreshMode.default()
     initialize: Optional[Initialize] = Initialize.default()
+    row_access_policy: Optional[str] = None
+    table_tag: Optional[str] = None
 
     @classmethod
     def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
@@ -76,6 +77,8 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
             "snowflake_warehouse": config_dict.get("snowflake_warehouse"),
             "refresh_mode": config_dict.get("refresh_mode"),
             "initialize": config_dict.get("initialize"),
+            "row_access_policy": config_dict.get("row_access_policy"),
+            "table_tag": config_dict.get("table_tag"),
         }
 
         return super().from_dict(kwargs_dict)  # type:ignore
@@ -91,6 +94,10 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
             "snowflake_warehouse": relation_config.config.extra.get(  # type:ignore
                 "snowflake_warehouse"
             ),
+            "row_access_policy": relation_config.config.extra.get(  # type:ignore
+                "row_access_policy"
+            ),
+            "table_tag": relation_config.config.extra.get("table_tag"),  # type:ignore
         }
 
         if refresh_mode := relation_config.config.extra.get("refresh_mode"):  # type:ignore
@@ -113,6 +120,8 @@ def parse_relation_results(cls, relation_results: RelationResults) -> Dict[str,
             "target_lag": dynamic_table.get("target_lag"),
             "snowflake_warehouse": dynamic_table.get("warehouse"),
             "refresh_mode": dynamic_table.get("refresh_mode"),
+            "row_access_policy": dynamic_table.get("row_access_policy"),
+            "table_tag": dynamic_table.get("table_tag"),
             # we don't get initialize since that's a one-time scheduler attribute, not a DT attribute
         }
 

dbt-snowflake/src/dbt/include/snowflake/macros/relations/dynamic_table/create.sql

@@ -31,14 +31,16 @@
         A valid DDL statement which will result in a new dynamic info schema table.
 -#}
 
-create dynamic table {{ relation }}
-    target_lag = '{{ dynamic_table.target_lag }}'
-    warehouse = {{ dynamic_table.snowflake_warehouse }}
-    {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
-    {{ optional('initialize', dynamic_table.initialize) }}
-    as (
-        {{ sql }}
-    )
+    create dynamic table {{ relation }}
+        target_lag = '{{ dynamic_table.target_lag }}'
+        warehouse = {{ dynamic_table.snowflake_warehouse }}
+        {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
+        {{ optional('initialize', dynamic_table.initialize) }}
+        {{ optional('with row access policy', dynamic_table.row_access_policy, equals_char='') }}
+        {{ optional('with tag', dynamic_table.table_tag, quote_char='(', equals_char='') }}
+        as (
+            {{ sql }}
+        )
 
 {%- endmacro %}
 
@@ -62,16 +64,18 @@ create dynamic table {{ relation }}
 
 {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
 
-create dynamic iceberg table {{ relation }}
-    target_lag = '{{ dynamic_table.target_lag }}'
-    warehouse = {{ dynamic_table.snowflake_warehouse }}
-    {{ optional('external_volume', catalog_relation.external_volume, "'") }}
-    catalog = 'SNOWFLAKE'  -- required, and always SNOWFLAKE for built-in Iceberg tables
-    base_location = '{{ catalog_relation.base_location }}'
-    {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
-    {{ optional('initialize', dynamic_table.initialize) }}
-    as (
-        {{ sql }}
-    )
+    create dynamic iceberg table {{ relation }}
+        target_lag = '{{ dynamic_table.target_lag }}'
+        warehouse = {{ dynamic_table.snowflake_warehouse }}
+        {{ optional('external_volume', catalog_relation.external_volume, "'") }}
+        catalog = 'SNOWFLAKE'  -- required, and always SNOWFLAKE for built-in Iceberg tables
+        base_location = '{{ catalog_relation.base_location }}'
+        {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
+        {{ optional('initialize', dynamic_table.initialize) }}
+        {{ optional('row_access_policy', dynamic_table.row_access_policy) }}
+        {{ optional('table_tag', dynamic_table.table_tag) }}
+        as (
+            {{ sql }}
+        )
 
 {%- endmacro %}

dbt-snowflake/src/dbt/include/snowflake/macros/relations/dynamic_table/replace.sql

@@ -49,6 +49,8 @@ create or replace dynamic table {{ relation }}
     warehouse = {{ dynamic_table.snowflake_warehouse }}
     {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
     {{ optional('initialize', dynamic_table.initialize) }}
+    {{ optional('with row access policy', dynamic_table.row_access_policy, equals_char='') }}
+    {{ optional('with tag', dynamic_table.table_tag, quote_char='(', equals_char='') }}
     as (
         {{ sql }}
     )
@@ -83,6 +85,8 @@ create or replace dynamic iceberg table {{ relation }}
     base_location = '{{ catalog_relation.base_location }}'
     {{ optional('refresh_mode', dynamic_table.refresh_mode) }}
     {{ optional('initialize', dynamic_table.initialize) }}
+    {{ optional('row_access_policy', dynamic_table.row_access_policy) }}
+    {{ optional('table_tag', dynamic_table.table_tag) }}
     as (
         {{ sql }}
     )

dbt-snowflake/src/dbt/include/snowflake/macros/relations/table/create.sql

@@ -73,6 +73,9 @@ as (
 
 {%- set copy_grants = config.get('copy_grants', default=false) -%}
 
+{%- set row_access_policy = config.get('row_access_policy', default=none) -%}
+{%- set table_tag = config.get('table_tag', default=none) -%}
+
 {%- set contract_config = config.get('contract') -%}
 {%- if contract_config.enforced -%}
     {{- get_assert_columns_equivalent(compiled_code) -}}
@@ -88,6 +91,8 @@ create or replace {{ transient }}table {{ relation }}
     {{ get_table_columns_and_constraints() }}
     {%- endif %}
     {% if copy_grants -%} copy grants {%- endif %}
+    {% if row_access_policy -%} with row access policy {{ row_access_policy }} {%- endif %}
+    {% if table_tag -%} with tag ({{ table_tag }}) {%- endif %}
     as (
         {%- if catalog_relation.cluster_by is not none -%}
         select * from (
@@ -126,6 +131,9 @@ alter table {{ relation }} resume recluster;
 
 {%- set copy_grants = config.get('copy_grants', default=false) -%}
 
+{%- set row_access_policy = config.get('row_access_policy', default=none) -%}
+{%- set table_tag = config.get('table_tag', default=none) -%}
+
 {%- set contract_config = config.get('contract') -%}
 {%- if contract_config.enforced -%}
     {{- get_assert_columns_equivalent(compiled_code) -}}
@@ -142,6 +150,8 @@ create or replace iceberg table {{ relation }}
     {{ optional('external_volume', catalog_relation.external_volume, "'") }}
     catalog = 'SNOWFLAKE'  -- required, and always SNOWFLAKE for built-in Iceberg tables
     base_location = '{{ catalog_relation.base_location }}'
+    {% if row_access_policy -%} with row access policy {{ row_access_policy }} {%- endif %}
+    {% if table_tag -%} with tag ({{ table_tag }}) {%- endif %}
     {% if copy_grants -%} copy grants {%- endif %}
 as (
     {%- if catalog_relation.cluster_by is not none -%}

dbt-snowflake/src/dbt/include/snowflake/macros/relations/view/create.sql

@@ -1,6 +1,8 @@
 {% macro snowflake__create_view_as_with_temp_flag(relation, sql, is_temporary=False) -%}
   {%- set secure = config.get('secure', default=false) -%}
   {%- set copy_grants = config.get('copy_grants', default=false) -%}
+  {%- set row_access_policy = config.get('row_access_policy', default=none) -%}
+  {%- set table_tag = config.get('table_tag', default=none) -%}
   {%- set sql_header = config.get('sql_header', none) -%}
 
   {{ sql_header if sql_header is not none }}
@@ -19,7 +21,10 @@
   {%- if contract_config.enforced -%}
     {{ get_assert_columns_equivalent(sql) }}
   {%- endif %}
-  {% if copy_grants -%} copy grants {%- endif %} as (
+  {% if copy_grants -%} copy grants {%- endif %}
+  {% if row_access_policy -%} with row access policy {{ row_access_policy }} {%- endif %}
+  {% if table_tag -%} with tag ({{ table_tag }}) {%- endif %}
+  as (
     {{ sql }}
   );
 {% endmacro %}