Add get_identifier to hive source in metadata ingestion

    The hive source currently does not include the catalog name
    in the schema name. This is due to the SQLAlchemy
    reflection that pyhive performs. Without the catalog
    name, multiple hive sources could have collisions
    with the urns generated. This function will add
    the database name where applicable and solve this problem.

metadata-ingestion/setup.py

@@ -126,6 +126,7 @@ def get_long_description():
             "ldap",
             "looker",
             "glue",
+            "hive",
             "datahub-kafka",
             "datahub-rest",
             # airflow is added below

metadata-ingestion/src/datahub/ingestion/source/hive.py

@@ -22,6 +22,12 @@ class HiveConfig(BasicSQLAlchemyConfig):
     # defaults
     scheme = "hive"
 
+    def get_identifier(self, schema: str, table: str) -> str:
+        regular = f"{schema}.{table}"
+        if self.database:
+            return f"{self.database}.{regular}"
+        return regular
+
 
 class HiveSource(SQLAlchemySource):
     def __init__(self, config, ctx):

metadata-ingestion/tests/unit/test_hive_source.py

@@ -0,0 +1,23 @@
+import unittest
+
+from datahub.ingestion.source.hive import HiveConfig
+
+
+class HiveSinkTest(unittest.TestCase):
+    def test_hive_configuration_get_indentifier_with_database(self):
+        test_db_name = "test_database"
+        test_schema_name = "test_schema"
+        test_table_name = "test_table"
+        config_dict = {
+            "username": "test",
+            "password": "test",
+            "host_port": "test:80",
+            "database": test_db_name,
+            "scheme": "hive+https",
+        }
+        hive_config = HiveConfig.parse_obj(config_dict)
+        expected_output = f"{test_db_name}.{test_schema_name}.{test_table_name}"
+        output = hive_config.get_identifier(
+            schema=test_schema_name, table=test_table_name
+        )
+        assert output == expected_output