address review: guard column_masks to V2 only, add security test

- Add V2 guard to apply_column_masks in apply_config_changeset to
  prevent unmasked data exposure in V1 (CTAS writes data before masks
  can be applied, whereas V2 creates an empty table first)
- Add TestV1IncrementalColumnMasksNotApplied: overrides apply_column_masks
  to raise error, verifying it is never called in V1 incremental path
- Remove V1 column_tags/column_masks tests that are no longer applicable
- Keep TestV1IncrementalSkipConfigChanges for metadata skip validation

Co-authored-by: Isaac

Author: moomindani

dbt/include/databricks/macros/relations/table/alter.sql

@@ -30,7 +30,9 @@
       {% if constraints %}
         {{ apply_constraints(target_relation, constraints) }}
       {% endif %}
-      {% if column_masks %}
+      {#-- Column masks are only applied in V2 to avoid a window where data is unmasked (CTAS in V1
+           writes data before masks can be applied, whereas V2 creates an empty table first) --#}
+      {% if column_masks and adapter.behavior.use_materialization_v2 %}
         {{ apply_column_masks(target_relation, column_masks) }}
       {% endif %}
     {%- endif -%}

tests/functional/adapter/incremental/fixtures.py

@@ -1118,24 +1118,6 @@ def model(dbt, spark):
     return spark.createDataFrame(data, schema=['id', 'msg', 'color'])
 """
 
-v1_config_changes_sql = """
-{{ config(
-    materialized = 'incremental',
-    unique_key = 'id',
-    merge_update_columns = ['msg'],
-) }}
-
-{% if not is_incremental() %}
-
-select cast(1 as bigint) as id, 'hello' as msg, 'blue' as color
-
-{% else %}
-
-select cast(1 as bigint) as id, 'updated' as msg, 'blue' as color
-
-{% endif %}
-"""
-
 v1_skip_config_changes_sql = """
 {{ config(
     materialized = 'incremental',
@@ -1154,34 +1136,6 @@ def model(dbt, spark):
 {% endif %}
 """
 
-v1_column_tags_a = """
-version: 2
-
-models:
-  - name: v1_config_changes_sql
-    columns:
-        - name: id
-          databricks_tags:
-            pii: "false"
-        - name: msg
-        - name: color
-"""
-
-v1_column_tags_b = """
-version: 2
-
-models:
-  - name: v1_config_changes_sql
-    columns:
-        - name: id
-          databricks_tags:
-            pii: "true"
-        - name: msg
-          databricks_tags:
-            source: "app"
-        - name: color
-"""
-
 fail_if_metadata_fetched_macros = """
 {% macro fetch_tags(relation) %}
   {{ exceptions.raise_compiler_error("fetch_tags should not be called when incremental_apply_config_changes is false") }}
@@ -1208,6 +1162,33 @@ def model(dbt, spark):
 {% endmacro %}
 """
 
+fail_if_column_masks_applied_macro = """
+{% macro apply_column_masks(relation, column_masks) %}
+  {{ exceptions.raise_compiler_error("apply_column_masks should not be called in V1 — column masks must only be applied in V2 to prevent unmasked data exposure") }}
+{% endmacro %}
+"""
+
+v1_column_mask_model_sql = """
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'id',
+) }}
+
+select cast(1 as bigint) as id, 'hello' as name
+"""
+
+v1_column_mask_schema = """
+version: 2
+
+models:
+  - name: v1_column_mask_model_sql
+    columns:
+        - name: id
+        - name: name
+          column_mask:
+            function: full_mask
+"""
+
 warn_unenforced_override_sql = """
 select "abc" as id
 """

tests/functional/adapter/incremental/test_v1_incremental_config_changes.py

@@ -5,125 +5,64 @@
 
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestV1IncrementalColumnTags:
-    """Test that V1 incremental path applies column tag changes via process_config_changes."""
+class TestV1IncrementalSkipConfigChanges:
+    """Test that incremental_apply_config_changes=false skips metadata fetch queries in V1 path."""
 
     @pytest.fixture(scope="class")
     def models(self):
         return {
-            "v1_config_changes_sql.sql": fixtures.v1_config_changes_sql,
-            "schema.yml": fixtures.v1_column_tags_a,
-        }
-
-    def test_changing_column_tags(self, project):
-        # First run creates the table
-        util.run_dbt(["run"])
-
-        # Update column tags
-        util.write_file(fixtures.v1_column_tags_b, "models", "schema.yml")
-        util.run_dbt(["run"])
-
-        # Verify column tags were applied
-        results = project.run_sql(
-            f"""
-            select column_name, tag_name, tag_value
-            from `system`.`information_schema`.`column_tags`
-            where schema_name = '{project.test_schema}'
-            and table_name = 'v1_config_changes_sql'
-            order by column_name, tag_name
-            """,
-            fetch="all",
-        )
-
-        tags_dict = {}
-        for row in results:
-            col = row.column_name
-            if col not in tags_dict:
-                tags_dict[col] = {}
-            tags_dict[col][row.tag_name] = row.tag_value
-
-        # Verify expected final state
-        expected_tags = {
-            "id": {"pii": "true"},
-            "msg": {"source": "app"},
+            "v1_skip_config_changes_sql.sql": fixtures.v1_skip_config_changes_sql,
         }
-        assert tags_dict == expected_tags
-
-
-@pytest.mark.skip_profile("databricks_cluster")
-class TestV1IncrementalColumnMasks:
-    """Test that V1 incremental path applies column mask changes via process_config_changes."""
 
     @pytest.fixture(scope="class")
-    def models(self):
+    def macros(self):
         return {
-            "column_mask_sql.sql": fixtures.column_mask_sql,
-            "schema.yml": fixtures.column_mask_base,
+            "fail_if_metadata_fetched.sql": fixtures.fail_if_metadata_fetched_macros,
         }
 
-    def test_changing_column_masks(self, project):
-        # Create mask functions
-        project.run_sql(
-            f"""
-            CREATE OR REPLACE FUNCTION
-                {project.database}.{project.test_schema}.full_mask(password STRING)
-            RETURNS STRING
-            RETURN '*****';
-            """
-        )
-        project.run_sql(
-            f"""
-            CREATE OR REPLACE FUNCTION
-                {project.database}.{project.test_schema}.email_mask(value STRING)
-            RETURNS STRING
-            RETURN CONCAT(
-                REPEAT('*', POSITION('@' IN value) - 1),
-                SUBSTR(value, POSITION('@' IN value))
-            );
-            """
-        )
-
-        # First run with initial masks
+    def test_incremental_run_skips_metadata_queries(self, project):
+        # First run creates the table
         util.run_dbt(["run"])
-        masks = project.run_sql(
-            "SELECT id, name, email, password FROM column_mask_sql",
-            fetch="all",
-        )
-        assert len(masks) == 1
-        assert masks[0][1] == "*****"  # name (masked)
-        assert masks[0][3] == "password123"  # password (unmasked)
-
-        # Update masks and verify changes
-        util.write_file(fixtures.column_mask_valid_mask_updates, "models", "schema.yml")
+        # Second run exercises the incremental merge path.
+        # If metadata fetch macros are called, they will raise errors and the run will fail.
         util.run_dbt(["run"])
 
-        result = project.run_sql(
-            "SELECT id, name, email, password FROM column_mask_sql", fetch="all"
-        )
-        assert len(result) == 1
-        assert result[0][1] == "hello"  # name (unmasked)
-        assert result[0][3] == "*****"  # password (masked)
-
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestV1IncrementalSkipConfigChanges:
-    """Test that incremental_apply_config_changes=false skips metadata fetch queries."""
+class TestV1IncrementalColumnMasksNotApplied:
+    """Test that column masks are NOT applied in V1 incremental path.
+
+    Column masks must only be applied in V2 where the empty table is created before data
+    arrives. In V1 (CTAS), data is written immediately, so applying masks after the fact
+    would leave a window where data is unmasked — a security/privacy vulnerability.
+    """
 
     @pytest.fixture(scope="class")
     def models(self):
         return {
-            "v1_skip_config_changes_sql.sql": fixtures.v1_skip_config_changes_sql,
+            "v1_column_mask_model_sql.sql": fixtures.v1_column_mask_model_sql,
+            "schema.yml": fixtures.v1_column_mask_schema,
         }
 
     @pytest.fixture(scope="class")
     def macros(self):
         return {
-            "fail_if_metadata_fetched.sql": fixtures.fail_if_metadata_fetched_macros,
+            "fail_if_column_masks_applied.sql": fixtures.fail_if_column_masks_applied_macro,
         }
 
-    def test_incremental_run_skips_metadata_queries(self, project):
+    def test_column_masks_not_applied_in_v1(self, project):
+        # Create the mask function so the model config is valid
+        project.run_sql(
+            f"""
+            CREATE OR REPLACE FUNCTION
+                {project.database}.{project.test_schema}.full_mask(val STRING)
+            RETURNS STRING
+            RETURN '*****';
+            """
+        )
+
         # First run creates the table
         util.run_dbt(["run"])
-        # Second run exercises the incremental merge path.
-        # If metadata fetch macros are called, they will raise errors and the run will fail.
+        # Second run exercises the incremental merge path with config change detection.
+        # If apply_column_masks is called, the overridden macro raises an error.
         util.run_dbt(["run"])