feat(superset): add an input so it's possible to ingest tables

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

@@ -105,6 +105,11 @@ class SupersetConfig(StatefulIngestionConfigBase, ConfigModel):
         description="Can be used to change mapping for database names in superset to what you have in datahub",
     )
 
+    reserved_character_replacement: str = Field(
+        default=None,
+        description=f"Can be used to replace reserved characters {RESERVED_CHARACTERS} in tables name so they can be ingested",
+    )
+
     @validator("connect_uri", "display_uri")
     def remove_trailing_slash(cls, v):
         return config_clean.remove_trailing_slashes(v)
@@ -219,10 +224,19 @@ def get_datasource_urn_from_id(self, datasource_id):
         schema_name = dataset_response.get("result", {}).get("schema")
         table_name = dataset_response.get("result", {}).get("table_name")
         if RESERVED_CHARACTERS.intersection(table_name):
-            logger.warning(
-                f"Failed to ingest following dataset due to a reserved character {RESERVED_CHARACTERS} in table name: {schema_name}.{table_name}"
+            if not self.config.reserved_character_replacement:
+                logger.warning(
+                    f"Failed to ingest following dataset due to a reserved character {RESERVED_CHARACTERS} in table name: {schema_name}.{table_name}."
+                    "Consider using 'reserved_character_replacement' argument to ingest dataset"
+                )
+                return None
+
+            table_name = re.sub(
+                rf"[{''.join(RESERVED_CHARACTERS)}]",
+                self.config.RESERVED_CHARACTERS,
+                table_name,
             )
-            return None
+
         database_id = dataset_response.get("result", {}).get("database", {}).get("id")
         database_name = (
             dataset_response.get("result", {}).get("database", {}).get("database_name")