feat(ingest/hive): identify partition columns in hive tables

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

@@ -777,6 +777,7 @@ def get_schema_fields_for_column(
             column,
             inspector,
             pk_constraints,
+            partition_keys=partition_keys,
         )
 
         if self._COMPLEX_TYPE.match(fields[0].nativeDataType) and isinstance(
@@ -849,3 +850,15 @@ def _process_view(
                 default_db=default_db,
                 default_schema=default_schema,
             )
+
+    def get_partitions(
+        self, inspector: Inspector, schema: str, table: str
+    ) -> Optional[List[str]]:
+        partition_columns: List[dict] = inspector.get_indexes(
+            table_name=table, schema=schema
+        )
+        for partition_column in partition_columns:
+            if partition_column.get("column_names"):
+                return partition_column.get("column_names")
+
+        return []

metadata-ingestion/tests/integration/hive/hive_mces_all_db_golden.json

@@ -1255,7 +1255,8 @@
                                 },
                                 "nativeDataType": "string",
                                 "recursive": false,
-                                "isPartOfKey": false
+                                "isPartOfKey": false,
+                                "isPartitioningKey": true
                             }
                         ]
                     }

metadata-ingestion/tests/integration/hive/hive_mces_golden.json

@@ -1255,7 +1255,8 @@
                                 },
                                 "nativeDataType": "string",
                                 "recursive": false,
-                                "isPartOfKey": false
+                                "isPartOfKey": false,
+                                "isPartitioningKey": true
                             }
                         ]
                     }