fix(hive): quote identifiers for DESCRIBE partition parsing

Author: SaaiAravindhRaja

ingestion/src/metadata/ingestion/source/database/hive/metadata.py

@@ -172,7 +172,9 @@ def get_schema_definition(  # pylint: disable=unused-argument
         return None
 
     def _build_metastore_partition_query(self, drivername: str) -> str:
-        q = lambda name: f'"{name}"' if drivername == "hive+postgres" else name  # noqa: E731
+        q = (
+            lambda name: f'"{name}"' if drivername == "hive+postgres" else name
+        )  # noqa: E731
         return (
             f"SELECT pk.{q('PKEY_NAME')}"
             f" FROM {q('PARTITION_KEYS')} pk"
@@ -197,8 +199,11 @@ def _get_partition_keys_from_describe(
     ) -> List[str]:
         partition_keys: List[str] = []
         in_partition_section = False
+        identifier_preparer = self.engine.dialect.identifier_preparer
+        quoted_schema_name = identifier_preparer.quote_identifier(schema_name)
+        quoted_table_name = identifier_preparer.quote_identifier(table_name)
         rows = self.connection.execute(
-            text(f"DESCRIBE FORMATTED `{schema_name}`.`{table_name}`")
+            text(f"DESCRIBE FORMATTED {quoted_schema_name}.{quoted_table_name}")
         )
         for row in rows:
             col_name = row[0].strip() if row[0] else ""

ingestion/tests/unit/topology/database/test_hive.py

@@ -360,15 +360,15 @@ def __init__(
         self.thread_id = self.hive.context.get_current_thread_id()
         self.hive._inspector_map[self.thread_id] = types.SimpleNamespace()
 
-        self.hive._inspector_map[self.thread_id].get_pk_constraint = (
-            lambda table_name, schema_name: []
-        )
-        self.hive._inspector_map[self.thread_id].get_unique_constraints = (
-            lambda table_name, schema_name: []
-        )
-        self.hive._inspector_map[self.thread_id].get_foreign_keys = (
-            lambda table_name, schema_name: []
-        )
+        self.hive._inspector_map[
+            self.thread_id
+        ].get_pk_constraint = lambda table_name, schema_name: []
+        self.hive._inspector_map[
+            self.thread_id
+        ].get_unique_constraints = lambda table_name, schema_name: []
+        self.hive._inspector_map[
+            self.thread_id
+        ].get_foreign_keys = lambda table_name, schema_name: []
 
     def test_yield_database(self):
         assert EXPECTED_DATABASE == [