fix(db2): implement grants via GrantsFromInfoSchemaMixin + role-based test infrastructure

db2.py:
- Add GrantsFromInfoSchemaMixin to Db2EngineAdapter — provides _get_current_grants_config,
  _apply_grants_config_expr, _revoke_grants_config_expr via INFORMATION_SCHEMA.table_privileges
- Set CURRENT_USER_OR_ROLE_EXPRESSION to CURRENT USER (Db2 special register)
- Add _grant_object_kind() returning 'TABLE' (Db2 GRANT requires the TABLE keyword)

Without the mixin, SUPPORTS_GRANTS=True with no implementations caused
NotImplementedError on every grant test method call.

__init__.py:
- Add db2 case to _get_create_user_or_role(): CREATE ROLE (Db2 LUW uses OS-level
  users for auth; roles work for GRANT/REVOKE testing without OS user setup)
- Add db2 to _cleanup_user_or_role(): DROP ROLE IF EXISTS (same as Snowflake)

Author: iawanish

sqlmesh/core/engine_adapter/db2.py

@@ -9,7 +9,10 @@
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 
 from sqlmesh.core.engine_adapter.base import EngineAdapter, _get_data_object_cache_key
-from sqlmesh.core.engine_adapter.mixins import PandasNativeFetchDFSupportMixin
+from sqlmesh.core.engine_adapter.mixins import (
+    GrantsFromInfoSchemaMixin,
+    PandasNativeFetchDFSupportMixin,
+)
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
     CommentCreationTable,
@@ -43,13 +46,16 @@ def is_db2_error(exception: Exception, error_code: str) -> bool:
 
 @set_catalog()
 class Db2EngineAdapter(
+    GrantsFromInfoSchemaMixin,
     PandasNativeFetchDFSupportMixin,
     EngineAdapter,
 ):
     DIALECT = "db2"
     SUPPORTS_INDEXES = True
     SUPPORTS_REPLACE_TABLE = False
     SUPPORTS_GRANTS = True
+    # Db2 uses CURRENT USER special register to identify the grantor.
+    CURRENT_USER_OR_ROLE_EXPRESSION: exp.Expr = exp.column("CURRENT USER")
     COMMENT_CREATION_TABLE = CommentCreationTable.COMMENT_COMMAND_ONLY
     COMMENT_CREATION_VIEW = CommentCreationView.COMMENT_COMMAND_ONLY
     SUPPORTS_QUERY_EXECUTION_TRACKING = True
@@ -324,6 +330,11 @@ def _db2_type_to_sqlglot(self, db2_type: str, length: int, scale: int) -> exp.Da
         sqlglot_type = type_mapping.get(db2_type, f"VARCHAR({length})")
         return exp.DataType.build(sqlglot_type, dialect="db2")
 
+    @staticmethod
+    def _grant_object_kind(table_type: DataObjectType) -> t.Optional[str]:
+        """Db2 GRANT/REVOKE requires TABLE keyword for tables and views."""
+        return "TABLE"
+
     @property
     def catalog_support(self) -> CatalogSupport:
         return CatalogSupport.SINGLE_CATALOG_ONLY

tests/core/engine_adapter/integration/__init__.py

@@ -818,6 +818,10 @@ def _get_create_user_or_role(
             project_id = self.engine_adapter.get_current_catalog()
             service_account = f"sqlmesh-test-{role_name}@{project_id}.iam.gserviceaccount.com"
             return f"serviceAccount:{service_account}", None
+        if self.dialect == "db2":
+            # Db2 LUW uses OS-level users for authentication, but database roles
+            # work for GRANT/REVOKE testing without requiring OS user setup.
+            return username, f"CREATE ROLE {username}"
         raise ValueError(f"User creation not supported for dialect: {self.dialect}")
 
     def _create_user_or_role(self, username: str, password: t.Optional[str] = None) -> str:
@@ -883,7 +887,7 @@ def _cleanup_user_or_role(self, user_name: str) -> None:
                 """)
                 self.engine_adapter.execute(f'DROP OWNED BY "{user_name}"')
                 self.engine_adapter.execute(f'DROP USER IF EXISTS "{user_name}"')
-            elif self.dialect == "snowflake":
+            elif self.dialect in ["snowflake", "db2"]:
                 self.engine_adapter.execute(f"DROP ROLE IF EXISTS {user_name}")
             elif self.dialect in ["databricks", "bigquery"]:
                 # For Databricks and BigQuery, we use pre-created accounts that should not be deleted