fix: handle grant column names case-insensitively

CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Fixes
 
+- Handle `SHOW GRANTS` result columns case-insensitively so grant reconciliation does not crash when connectors return lowercase names (thanks @TangoEnSkai!) ([#1650](https://github.com/databricks/dbt-databricks/pull/1650) resolves [#782](https://github.com/databricks/dbt-databricks/issues/782))
 - Replace an existing table or view with a metric view using backup-and-create instead of `CREATE OR REPLACE VIEW ... WITH METRICS` ([#1640](https://github.com/databricks/dbt-databricks/pull/1640) resolves [#1639](https://github.com/databricks/dbt-databricks/issues/1639))
 - Interpolate lazily-formatted `databricks.sql` log records when mirroring them into dbt logs ([#1642](https://github.com/databricks/dbt-databricks/pull/1642) resolves [#1637](https://github.com/databricks/dbt-databricks/issues/1637))
 

dbt/adapters/databricks/impl.py

@@ -317,6 +317,20 @@ def _has_dbr_capability_parse(self, capability_name: str) -> bool:
     def _v2_to_v1_type(self, catalog_type: str) -> str:
         return self._V2_TO_V1_TYPE.get(catalog_type, catalog_type)
 
+    def standardize_grants_dict(self, grants_table: "Table") -> dict[str, list[str]]:
+        column_names = {name.lower(): name for name in grants_table.column_names}
+        grants_dict: dict[str, list[str]] = {}
+
+        for row in grants_table:
+            grantee = row[column_names["principal"]]
+            privilege = row[column_names["actiontype"]]
+            object_type = row[column_names["objecttype"]]
+
+            if object_type == "TABLE" and privilege != "OWN":
+                grants_dict.setdefault(privilege, []).append(grantee)
+
+        return grants_dict
+
     @property
     def _behavior_flags(self) -> list[BehaviorFlag]:
         return [

tests/functional/adapter/grants/fixtures.py

@@ -0,0 +1,13 @@
+granted_table_sql = """
+{{ config(materialized='table') }}
+select 1 as id
+"""
+
+granted_table_schema_yml = """
+version: 2
+models:
+  - name: granted_table
+    config:
+      grants:
+        select: ["account users"]
+"""

tests/functional/adapter/grants/test_grants.py

@@ -4,6 +4,10 @@
 from dbt.tests.adapter.grants.test_model_grants import BaseModelGrants
 from dbt.tests.adapter.grants.test_seed_grants import BaseSeedGrants
 from dbt.tests.adapter.grants.test_snapshot_grants import BaseSnapshotGrants
+from dbt.tests.util import run_dbt
+
+from tests.functional.adapter.fixtures import RerunSafeMixin
+from tests.functional.adapter.grants import fixtures
 
 
 @pytest.mark.skip(reason="DECO team must provide DBT_TEST_USER_1/2/3 before we re-enable")
@@ -49,3 +53,29 @@ def grantee_does_not_exist_error(self):
 
     def privilege_does_not_exist_error(self):
         return "INVALID_PARAMETER_VALUE"
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableGrantReconciliation(RerunSafeMixin):
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("granted_table",)
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "granted_table.sql": fixtures.granted_table_sql,
+            "schema.yml": fixtures.granted_table_schema_yml,
+        }
+
+    def test_second_run_reconciles_show_grants(self, project):
+        run_dbt(["run"])
+        run_dbt(["run"])
+
+        rows = project.run_sql(
+            "select grantee, privilege_type from {database}.information_schema.table_privileges "
+            "where table_schema = '{schema}' and table_name = 'granted_table' "
+            "and inherited_from = 'NONE'",
+            fetch="all",
+        )
+        assert ("account users", "SELECT") in {(row[0], row[1]) for row in rows}

tests/unit/test_adapter.py

@@ -135,6 +135,30 @@ def _stub_spog_probe(self):
         ):
             yield
 
+    @pytest.mark.parametrize(
+        "column_names",
+        [
+            ["Principal", "ActionType", "ObjectType", "ObjectKey"],
+            ["principal", "actiontype", "objecttype", "objectkey"],
+            ["principal", "actionType", "objectType", "objectKey"],
+        ],
+    )
+    def test_standardize_grants_dict_ignores_column_name_case(self, column_names):
+        grants_table = agate.Table(
+            [
+                ["analysts", "SELECT", "TABLE", "catalog.schema.model"],
+                ["engineers", "SELECT", "TABLE", "catalog.schema.model"],
+                ["owner", "OWN", "TABLE", "catalog.schema.model"],
+                ["catalog_user", "USE CATALOG", "CATALOG", "catalog"],
+            ],
+            column_names=column_names,
+        )
+        adapter = DatabricksAdapter(self._get_config(), get_context("spawn"))
+
+        assert adapter.standardize_grants_dict(grants_table) == {
+            "SELECT": ["analysts", "engineers"]
+        }
+
     def test_two_catalog_settings(self):
         with pytest.raises(DbtConfigError) as excinfo:
             self._get_config(