Warn when documented columns are missing from the relation in persist_docs

When persist_docs.columns is enabled, columns documented in a model's
schema.yml but absent from the materialized relation were silently
skipped by get_persist_doc_columns. Emit a warning naming those columns
so users can catch typos and stale documentation. The columns are still
filtered out (no behavior change to the comments that get applied).

Ports the behavior added upstream in dbt-adapters#1684 (issue #1690).

Author: r-jais

CHANGELOG.md

@@ -5,6 +5,8 @@
 - Support `catalog_database` in v2 catalogs.yml to route Unity catalog models to a physical catalog independent of the dbt catalog name (requires `dbt-core>=1.12` and `dbt-adapters>=1.24.4`). ([#1590](https://github.com/databricks/dbt-databricks/pull/1590))
 
 ### Fixes
+
+- Warn when a column documented in a model's `schema.yml` is absent from the relation while applying column comments, instead of silently skipping it — surfaces typos and stale column documentation. Covers the column-comment comparison paths (V1 `get_persist_doc_columns` and V2 `ColumnCommentsConfig.get_diff`); create-time inline comments are not yet covered. Ports the behavior added in dbt-adapters ([dbt-adapters#1684](https://github.com/dbt-labs/dbt-adapters/pull/1684) closes [dbt-adapters#1690](https://github.com/dbt-labs/dbt-adapters/issues/1690)) ([#1563](https://github.com/databricks/dbt-databricks/pull/1563)).
 - Support `dbt clone` and rebuilds over an existing shallow clone ([#1592](https://github.com/databricks/dbt-databricks/pull/1592) resolves [#1165](https://github.com/databricks/dbt-databricks/issues/1165))
 - Fix managed Iceberg Python models failing with `MANAGED_TABLE_FORMAT` by emitting `.format("iceberg")` instead of the `parquet` sentinel from `resolve_file_format` (thanks @Divya-Kovvuru-0802!) ([#1593](https://github.com/databricks/dbt-databricks/pull/1593) resolves [#1591](https://github.com/databricks/dbt-databricks/issues/1591))
 - Quote generated column identifiers in incremental strategies so non-ASCII column names no longer fail on subsequent runs ([#1595](https://github.com/databricks/dbt-databricks/pull/1595) resolves [#1594](https://github.com/databricks/dbt-databricks/issues/1594))

dbt/adapters/databricks/impl.py

@@ -20,6 +20,7 @@
 from dbt.adapters.catalogs import CatalogRelation
 from dbt.adapters.contracts.connection import AdapterResponse, Connection
 from dbt.adapters.contracts.relation import RelationConfig, RelationType
+from dbt.adapters.events.types import AdapterEventWarning
 from dbt.adapters.relation_configs import RelationResults
 from dbt.adapters.spark.impl import (
     DESCRIBE_TABLE_EXTENDED_MACRO_NAME,
@@ -31,6 +32,7 @@
 )
 from dbt_common.behavior_flags import BehaviorFlag
 from dbt_common.contracts.config.base import BaseConfig, MergeBehavior
+from dbt_common.events.functions import warn_or_error
 from dbt_common.exceptions import DbtConfigError, DbtInternalError, DbtRuntimeError
 from dbt_common.record import auto_record_function, record_function
 from dbt_common.utils import executor
@@ -1008,6 +1010,25 @@ def get_persist_doc_columns(
         # Create a case-insensitive lookup for column names
         columns_lower = {k.lower(): k for k in columns.keys()}
 
+        # Warn about columns that are documented in the model's schema but are not present in the
+        # relation. These are silently skipped below (rather than erroring on the alter), so surface
+        # them to the user to catch typos and stale documentation.
+        existing_lower = {column.column.lower() for column in existing_columns}
+        missing = [
+            original_name
+            for name_lower, original_name in columns_lower.items()
+            if name_lower not in existing_lower
+        ]
+        if missing:
+            warn_or_error(
+                AdapterEventWarning(
+                    base_msg=(
+                        "The following columns are specified in the schema but are not present "
+                        "in the database and will be skipped: " + ", ".join(missing)
+                    )
+                )
+            )
+
         for column in existing_columns:
             name = column.column
             # Use case-insensitive comparison for column names

tests/unit/test_adapter.py

@@ -1182,6 +1182,31 @@ def test_get_persist_doc_columns_case_mismatch_no_update_needed(self, adapter):
         # No update needed since comments match
         assert result == {}
 
+    @patch("dbt.adapters.databricks.impl.warn_or_error")
+    def test_get_persist_doc_columns_warns_on_missing_column(self, mock_warn, adapter):
+        """Documented columns absent from the relation are warned about and skipped."""
+        existing = [self.create_column("col1", "comment1")]
+        column_dict = {
+            "col1": {"name": "col1", "description": "new comment"},
+            "col2": {"name": "col2", "description": "comment for missing column"},
+        }
+        result = adapter.get_persist_doc_columns(existing, column_dict)
+        # The missing column is filtered out; only the existing column is returned.
+        assert result == {"col1": {"name": "col1", "description": "new comment"}}
+        # A warning is emitted naming the missing column.
+        mock_warn.assert_called_once()
+        warned_event = mock_warn.call_args.args[0]
+        assert "col2" in warned_event.base_msg
+        assert "col1" not in warned_event.base_msg
+
+    @patch("dbt.adapters.databricks.impl.warn_or_error")
+    def test_get_persist_doc_columns_no_warning_when_all_present(self, mock_warn, adapter):
+        """No warning is emitted when every documented column exists (case-insensitively)."""
+        existing = [self.create_column("Account_ID", "")]
+        column_dict = {"account_id": {"name": "account_id", "description": "Account ID column"}}
+        adapter.get_persist_doc_columns(existing, column_dict)
+        mock_warn.assert_not_called()
+
 
 class TestGetColumnsByDbrVersion(DatabricksAdapterBase):
     @pytest.fixture