feat(ingest/GX): add urn lowercasing option for GX assertions (datahu…

…b-project#7472)

metadata-ingestion/integration_docs/great-expectations.md

@@ -49,6 +49,7 @@ This integration does not support
     - `retry_max_times` (optional): Maximum times to retry if HTTP request fails. The delay between retries is increased exponentially.
     - `extra_headers` (optional): Extra headers which will be added to the datahub request.
     - `parse_table_names_from_sql` (defaults to false): The integration can use an SQL parser to try to parse the datasets being asserted. This parsing is disabled by default, but can be enabled by setting `parse_table_names_from_sql: True`.  The parser is based on the [`sqllineage`](https://pypi.org/project/sqllineage/) package.
+    - `convert_urns_to_lowercase` (optional): Whether to convert dataset urns to lowercase.
     
 ## Debugging
 Set environment variable `DATAHUB_DEBUG` (default `false`) to `true` to enable debug logging for `DataHubValidationAction`.

metadata-ingestion/src/datahub/integrations/great_expectations/action.py

@@ -87,6 +87,7 @@ def __init__(
         extra_headers: Optional[Dict[str, str]] = None,
         exclude_dbname: Optional[bool] = None,
         parse_table_names_from_sql: bool = False,
+        convert_urns_to_lowercase: bool = False,
     ):
         super().__init__(data_context)
         self.server_url = server_url
@@ -101,6 +102,7 @@ def __init__(
         self.extra_headers = extra_headers
         self.exclude_dbname = exclude_dbname
         self.parse_table_names_from_sql = parse_table_names_from_sql
+        self.convert_urns_to_lowercase = convert_urns_to_lowercase
 
     def _run(
         self,
@@ -593,6 +595,7 @@ def get_dataset_partitions(self, batch_identifier, data_asset):
                     ),
                     self.exclude_dbname,
                     self.platform_alias,
+                    self.convert_urns_to_lowercase,
                 )
                 batchSpec = BatchSpec(
                     nativeBatchId=batch_identifier,
@@ -661,6 +664,7 @@ def get_dataset_partitions(self, batch_identifier, data_asset):
                         ),
                         self.exclude_dbname,
                         self.platform_alias,
+                        self.convert_urns_to_lowercase,
                     )
                     dataset_partitions.append(
                         {
@@ -703,6 +707,7 @@ def make_dataset_urn_from_sqlalchemy_uri(
     platform_instance=None,
     exclude_dbname=None,
     platform_alias=None,
+    convert_urns_to_lowercase=False,
 ):
     data_platform = get_platform_from_sqlalchemy_uri(str(sqlalchemy_uri))
     url_instance = make_url(sqlalchemy_uri)
@@ -777,6 +782,9 @@ def make_dataset_urn_from_sqlalchemy_uri(
 
     dataset_name = f"{schema_name}.{table_name}"
 
+    if convert_urns_to_lowercase:
+        dataset_name = dataset_name.lower()
+
     dataset_urn = builder.make_dataset_urn_with_platform_instance(
         platform=data_platform if platform_alias is None else platform_alias,
         name=dataset_name,