feat: column-level Dynamic Data Masking (masked_with / masks)

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 #### Features
 
+- Add column-level Dynamic Data Masking (DDM). Declare masks via a first-class `masked_with:` column property in schema YAML or a model-level `masks` ({column: function}) config (which merges key-wise across `dbt_project.yml`, `.yml` and in-file `config()`), and the adapter (re)applies them on every build so they survive full-refresh rebuilds. Diffs against `sys.masked_columns` and emits only changed `ADD`/`MASKED WITH`/`DROP MASKED` DDL; column-level wins over model-level (warned); `masked_with: null` opts a column out of an inherited default; unknown columns are skipped with a warning; unmaskable column types error. Applies to `table`/`incremental`/`snapshot` base tables only (not views or seeds), and masks are applied before index creation so masked index-key columns work on fresh builds. Requires SQL Server 2016+.
 - Add Postgres-style `indexes` model config for tables, incrementals, seeds and snapshots, covering most `CREATE INDEX` options. [#535](https://github.com/dbt-msft/dbt-sqlserver/issues/535)
 - Index names are deterministic definition hashes (`dbt_idx_` prefix); creation is idempotent and unchanged definitions are never rebuilt.
 - Reconcile indexes against the config on incremental, DML-refresh and snapshot runs, applied as one atomic batch. Constraint-backing, legacy post-hook and `as_columnstore` indexes are never dropped.

README.md

@@ -216,6 +216,53 @@ You can also set it per model:
 {{ config(materialized="table", as_columnstore=false) }}
 ```
 
+### Dynamic Data Masking (`masked_with` / `masks`)
+
+The adapter can apply SQL Server [Dynamic Data Masking](https://learn.microsoft.com/en-us/sql/relational-databases/security/dynamic-data-masking) (DDM) to columns as part of the materialization, so masks are re-applied on every build and survive dbt's drop-and-recreate on a full refresh. A principal granted `SELECT` but not `UNMASK` then sees masked values instead of real data (dbt's own build principal, being `db_owner`, keeps `UNMASK` and reads real data). Requires **SQL Server 2016+**.
+
+There are two config surfaces, and you can use either or both:
+
+**Column-level `masked_with:`** — a first-class column property in schema YAML (like `data_type:` or `constraints:`), whose value is the masking-function string:
+
+```yaml
+# models/schema.yml
+version: 2
+models:
+  - name: core_patients
+    columns:
+      - name: surname
+        masked_with: "default()"
+      - name: nhs_number
+        masked_with: 'partial(0,"XXXXXXXXXX",0)'
+```
+
+**Model-level `masks`** — a `{column: function}` dict, settable in the in-file `{{ config() }}`, the model's `.yml` `config:` block, or a directory-wide default in `dbt_project.yml`. It merges key-wise across those levels (like `meta`), so a directory default and a per-model tweak combine rather than clobber:
+
+```sql
+{{ config(masks={'surname': "default()", 'nhs_number': 'partial(0,"XXXXXXXXXX",0)'}) }}
+```
+
+```yaml
+# dbt_project.yml — mask nhs_number on every model under datasets/ that has it
+models:
+  my_project:
+    datasets:
+      +masks: { nhs_number: "default()" }
+```
+
+Behaviour:
+
+- **Precedence:** when both surfaces target the same column, the column-level `masked_with` wins, and a warning naming the model, column and both functions is emitted (even when they agree). This is not something dbt itself ranks, so the rule is the adapter's: a column is more specific than a model.
+- **Opt out of an inherited default:** set `masked_with: null` on the column to remove a mask inherited from a directory/model-level `masks` entry.
+- **Validation:** a `masks` (or `masked_with`) entry naming a column that is not in the built relation is skipped with a warning (a likely typo or stale rename); the run does not fail.
+- **Unmaskable columns:** computed, `FILESTREAM`, sparse `COLUMN_SET` and `Always Encrypted` columns cannot carry a mask, and the run errors listing them rather than emitting DDL that fails.
+- **Views/ephemeral/seeds:** masks apply to base tables only (`table`, `incremental`, `snapshot`). Views inherit masking from their base tables and cannot carry a mask; **seeds are not currently masked**.
+- **Idempotent:** the adapter diffs the desired masks against `sys.masked_columns` and emits only the `ADD` / change / `DROP MASKED` statements that changed, so a persisted (incremental) re-run with no config change issues no DDL.
+
+**Indexes and masking.** SQL Server cannot *add* a mask to a column an index depends on (documented for all versions: `ALTER TABLE ALTER COLUMN … failed because one or more objects access this column`). The adapter avoids this on fresh builds by applying masks **before** it creates (rowstore) indexes — which is exactly Microsoft's documented workaround order (mask, then create the index). The default clustered columnstore index is unaffected (its columns are included, not key columns). On a **persisted** table (incremental/snapshot without full refresh), adding a *new* mask to a column that is already an index key errors with a message pointing to the drop-index → mask → recreate-index workaround.
+
+**Version notes.** All masking DDL the adapter emits (`ADD MASKED`, `MASKED WITH`, `DROP MASKED`) and the functions `default()`, `email()`, `random(a,b)` and `partial(...)` work on 2016+. The `datetime()` partial-date function and granular column/schema/table-scoped `UNMASK` are SQL Server 2022+ only; the adapter never emits them, but mask-function strings are passed through verbatim, so using a 2022-only function on an older server will be rejected by SQL Server.
+
 ## Contributing
 
 [![Unit tests](https://github.com/dbt-msft/dbt-sqlserver/actions/workflows/unit-tests.yml/badge.svg)](https://github.com/dbt-msft/dbt-sqlserver/actions/workflows/unit-tests.yml)

dbt/adapters/sqlserver/sqlserver_adapter.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 
 import agate
 import dbt_common.exceptions
@@ -29,6 +29,9 @@
 from dbt.adapters.sqlserver.sqlserver_column import SQLServerColumn, SQLServerColumnNative
 from dbt.adapters.sqlserver.sqlserver_configs import SQLServerConfigs
 from dbt.adapters.sqlserver.sqlserver_connections import SQLServerConnectionManager
+from dbt.adapters.sqlserver.sqlserver_mask import ColumnMask
+from dbt.adapters.sqlserver.sqlserver_mask import mask_changes as _mask_changes
+from dbt.adapters.sqlserver.sqlserver_mask import resolve_masks as _resolve_masks
 from dbt.adapters.sqlserver.sqlserver_relation import SQLServerRelation
 
 logger = AdapterLogger("SQLServer")
@@ -502,6 +505,64 @@ def index_changes(
             "warnings": warnings,
         }
 
+    @available
+    def resolve_masks(self, model: Any, model_masks: Optional[dict] = None) -> Dict[str, str]:
+        """Merge the column-level `masked_with` and model-level `masks` surfaces
+        into one `{column: function}` map for `apply_masks`.
+
+        `model` is the Jinja `model` dict (`node.to_dict()`), whose `columns`
+        carry any `masked_with` as a flattened key (an explicit `masked_with:
+        null` survives serialization as a present `None`, signalling opt-out).
+        `model_masks` is `config.get('masks')` — already surface-merged by dbt.
+        Precedence and conflict warnings are handled here; key existence is
+        validated later in the macro against the real relation.
+        """
+        model = model or {}
+        columns = model.get("columns") or {}
+        column_masks = []
+        for name, col in columns.items():
+            col = col or {}
+            column_masks.append(
+                ColumnMask(
+                    name=col.get("name", name),
+                    masked_with_present=("masked_with" in col),
+                    masked_with=col.get("masked_with"),
+                )
+            )
+        model_name = model.get("name") or model.get("alias") or "<unknown>"
+        mask_map, warnings = _resolve_masks(column_masks, model_masks, model_name)
+        for warning in warnings:
+            logger.warning(warning)
+        return mask_map
+
+    @available
+    def mask_changes(
+        self,
+        existing_masks: Any,
+        mask_config: Optional[dict],
+        index_key_columns: Any = None,
+        existing_columns: Any = None,
+    ) -> dict:
+        """Diff a resolved mask map against current `sys.masked_columns` state.
+
+        `existing_masks` is the agate table from `get_show_mask_sql` (columns
+        `name`, `masking_function`). Returns plain lists for jinja: `adds` /
+        `changes` (each `[column, function]`), `drops` (column names), `skipped`
+        (warnings for columns absent from the relation) and `errors` (an ADD onto
+        a current index-key column, which SQL Server rejects). The macro emits
+        DDL for adds/changes/drops, logs `skipped`, and raises on `errors`."""
+        rows = []
+        if existing_masks is not None:
+            column_names = existing_masks.column_names
+            for row in existing_masks.rows:
+                rows.append(dict(zip(column_names, row)))
+        return _mask_changes(
+            rows,
+            mask_config or {},
+            set(index_key_columns or []),
+            existing_columns=(list(existing_columns) if existing_columns is not None else None),
+        )
+
 
 COLUMNS_EQUAL_SQL = """
 with diff_count as (

dbt/adapters/sqlserver/sqlserver_configs.py

@@ -1,5 +1,7 @@
-from dataclasses import dataclass
-from typing import Any, Optional, Tuple
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, Tuple
+
+from dbt_common.contracts.config.base import MergeBehavior
 
 from dbt.adapters.protocol import AdapterConfig
 from dbt.adapters.sqlserver.relation_configs import SQLServerIndexConfig
@@ -14,3 +16,12 @@ class SQLServerConfigs(AdapterConfig):
     # false (default) | warn | true - how index reconciliation treats
     # droppable indexes dbt didn't create (YAML may supply bool or str)
     drop_unmanaged_indexes: Optional[Any] = False
+    # column-name -> DDM masking-function map for the model-level `masks`
+    # surface. MergeBehavior.Update makes it merge key-wise across the config
+    # chain (dbt_project.yml +masks defaults, .yml config, in-file config())
+    # the same way `meta` composes, rather than the default clobber — so a
+    # directory-level default and a per-model tweak combine instead of one
+    # replacing the whole dict.
+    masks: Optional[Dict[str, Any]] = field(
+        default_factory=dict, metadata=MergeBehavior.Update.meta()
+    )

dbt/adapters/sqlserver/sqlserver_mask.py

@@ -0,0 +1,199 @@
+"""Pure resolution + diff logic for Dynamic Data Masking (DDM).
+
+Kept free of any database or dbt-context dependency so it can be unit tested
+in isolation, mirroring ``relation_configs/index.py``. The adapter's
+``@available`` wrappers extract plain data from the model / ``sys.masked_columns``
+and delegate here; the Jinja ``sqlserver__apply_masks`` macro turns the diff
+into DDL.
+
+Two config surfaces feed one per-column mask map:
+
+* column-level ``masked_with:`` property (a value, or ``null`` to opt out), and
+* the model-level ``masks`` dict.
+
+Column-level wins over model-level, and any conflict is surfaced as a warning.
+Column identifiers are compared case-insensitively, matching SQL Server's
+default collation.
+"""
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Sequence, Set, Tuple
+
+
+@dataclass(frozen=True)
+class ColumnMask:
+    """A column's ``masked_with`` declaration, as parsed from schema YAML.
+
+    ``masked_with_present`` distinguishes "the key was written" from "absent":
+    an explicit ``masked_with: null`` (present with ``masked_with=None``) means
+    "no mask here", overriding any inherited model-level ``masks`` entry.
+    """
+
+    name: str
+    masked_with_present: bool
+    masked_with: Optional[str]
+
+
+def _normalize_name(name: str) -> str:
+    return name.strip().lower()
+
+
+def _pop_ci(mapping: Dict[str, str], name: str) -> Optional[str]:
+    """Remove a key matching ``name`` case-insensitively; return its value."""
+    target = _normalize_name(name)
+    for key in list(mapping):
+        if _normalize_name(key) == target:
+            return mapping.pop(key)
+    return None
+
+
+def _find_ci(mapping: Dict[str, str], name: str) -> Optional[str]:
+    """Return the existing key matching ``name`` case-insensitively, if any."""
+    target = _normalize_name(name)
+    for key in mapping:
+        if _normalize_name(key) == target:
+            return key
+    return None
+
+
+def resolve_masks(
+    column_masks: Sequence[ColumnMask],
+    model_masks: Optional[Dict[str, str]],
+    model_name: str,
+) -> Tuple[Dict[str, str], List[str]]:
+    """Merge the two config surfaces into one ``{column: function}`` map.
+
+    ``model_masks`` seeds the map (it is already surface-merged by dbt across
+    ``dbt_project.yml`` / ``.yml`` / in-file ``config()``). Each column-level
+    ``masked_with`` then overrides:
+
+    * a function value wins over any model-level entry for the same column and,
+      when both surfaces target that column, emits a conflict warning (even when
+      the two functions agree — a duplicate declaration is worth surfacing);
+    * an explicit ``null`` removes any inherited model-level entry ("opt out").
+
+    Key *existence* is not validated here — that happens in the macro against
+    the actually-built relation, which is authoritative and complete even when
+    the YAML ``columns:`` block is partial.
+    """
+    mask_map: Dict[str, str] = dict(model_masks or {})
+    warnings: List[str] = []
+
+    for cm in column_masks:
+        if not cm.masked_with_present:
+            continue
+
+        if cm.masked_with is None:
+            # explicit opt-out: drop any inherited model-level mask
+            _pop_ci(mask_map, cm.name)
+            continue
+
+        conflict_key = _find_ci(model_masks or {}, cm.name)
+        if conflict_key is not None:
+            warnings.append(
+                f"On model '{model_name}', column '{cm.name}' is masked by both a "
+                f"column-level `masked_with` ('{cm.masked_with}') and a model-level "
+                f"`masks` entry ('{(model_masks or {})[conflict_key]}'). The column-level "
+                f"function ('{cm.masked_with}') is applied (column-level wins)."
+            )
+
+        # column-level wins: drop any case-variant model-level key, then set
+        _pop_ci(mask_map, cm.name)
+        mask_map[cm.name] = cm.masked_with
+
+    return mask_map, warnings
+
+
+def _normalize_function(function: str) -> str:
+    """Canonicalise a masking-function string for comparison only.
+
+    ``sys.masked_columns.masking_function`` may store a function with different
+    internal spacing than the user wrote (e.g. ``partial(0, "X", 0)`` vs
+    ``partial(0,"X",0)``), so whitespace is stripped and case folded before
+    comparing desired against current. The user's original string is always the
+    one emitted in DDL.
+    """
+    return "".join(function.split()).lower()
+
+
+def mask_changes(
+    existing_masks: Sequence[Dict[str, str]],
+    desired: Dict[str, str],
+    index_key_columns: Set[str],
+    existing_columns: Optional[Sequence[str]] = None,
+) -> Dict[str, list]:
+    """Diff the desired mask map against current ``sys.masked_columns`` state.
+
+    ``existing_masks`` is a sequence of ``{"name", "masking_function"}`` rows.
+    ``existing_columns``, when given, is every column of the built relation; any
+    desired mask naming a column that is not present is skipped (a typo or a
+    stale rename) and reported in ``skipped`` rather than emitted as DDL that
+    would fail. Passing ``None`` disables that check.
+
+    Returns lists keyed ``adds`` / ``changes`` / ``drops`` / ``skipped`` /
+    ``errors``:
+
+    * ``adds`` – ``[(column, function)]`` masked in config, not yet in the DB
+      (``ALTER COLUMN ... ADD MASKED``);
+    * ``changes`` – ``[(column, function)]`` masked in the DB with a different
+      function (``ALTER COLUMN ... MASKED WITH``, no ``ADD``);
+    * ``drops`` – ``[column]`` masked in the DB but no longer in config
+      (``ALTER COLUMN ... DROP MASKED``);
+    * ``skipped`` – warning strings for desired columns absent from the relation;
+    * ``errors`` – ``ADD`` targeting a column that is currently an index key,
+      which SQL Server rejects with a dependency error (documented for all
+      versions 2016+). Only *adds* hit this: re-specifying the function on an
+      already-masked index-key column is fine. On fresh builds masks are applied
+      before indexes exist, so ``index_key_columns`` is empty there and this
+      never triggers; it only guards the persisted path where an index already
+      references the column.
+    """
+    existing_by_name = {
+        _normalize_name(row["name"]): row["masking_function"] for row in existing_masks
+    }
+    desired_by_name = {_normalize_name(col): (col, fn) for col, fn in desired.items()}
+    index_keys = {_normalize_name(c) for c in index_key_columns}
+    known_columns = (
+        {_normalize_name(c) for c in existing_columns} if existing_columns is not None else None
+    )
+
+    adds: List[Tuple[str, str]] = []
+    changes: List[Tuple[str, str]] = []
+    drops: List[str] = []
+    skipped: List[str] = []
+    errors: List[str] = []
+
+    for norm, (col, fn) in desired_by_name.items():
+        if known_columns is not None and norm not in known_columns:
+            skipped.append(
+                f"Column '{col}' is configured for masking but is not a column of "
+                f"the built relation; skipping. Check for a typo or a renamed column."
+            )
+            continue
+        if norm not in existing_by_name:
+            if norm in index_keys:
+                errors.append(
+                    f"Column '{col}' is configured for masking but is also an index "
+                    f"key column. SQL Server cannot add a mask to a column that an "
+                    f"index depends on. Remove the mask, or remove the column from the "
+                    f"index (drop the index, apply the mask, then recreate the index)."
+                )
+                continue
+            adds.append((col, fn))
+        elif _normalize_function(existing_by_name[norm]) != _normalize_function(fn):
+            changes.append((col, fn))
+
+    for norm, current_fn in existing_by_name.items():
+        if norm not in desired_by_name:
+            # preserve the DB's spelling of the column name for the DROP
+            drops.append(
+                next(row["name"] for row in existing_masks if _normalize_name(row["name"]) == norm)
+            )
+
+    return {
+        "adds": adds,
+        "changes": changes,
+        "drops": drops,
+        "skipped": skipped,
+        "errors": errors,
+    }