MINOR: Mysql Lineage Support Main (#18780)

* MINOR: Mysql Lineage Support Main

* fix test

* fix test

---------

Co-authored-by: Teddy <teddy.crepineau@gmail.com>

ingestion/src/metadata/ingestion/source/database/mysql/connection.py

@@ -38,6 +38,7 @@
     test_connection_db_schema_sources,
 )
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
+from metadata.ingestion.source.database.mysql.queries import MYSQL_TEST_GET_QUERIES
 from metadata.utils.constants import THREE_MIN
 
 
@@ -73,10 +74,14 @@ def test_connection(
     Test connection. This can be executed either as part
     of a metadata workflow or during an Automation Workflow
     """
+    queries = {
+        "GetQueries": MYSQL_TEST_GET_QUERIES,
+    }
     return test_connection_db_schema_sources(
         metadata=metadata,
         engine=engine,
         service_connection=service_connection,
         automation_workflow=automation_workflow,
         timeout_seconds=timeout_seconds,
+        queries=queries,
     )

ingestion/src/metadata/ingestion/source/database/mysql/lineage.py

@@ -9,38 +9,20 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 """
-Mysql lineage module
+MYSQL lineage module
 """
-from typing import Optional
-
-from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
-    MysqlConnection,
-)
-from metadata.generated.schema.metadataIngestion.workflow import (
-    Source as WorkflowSource,
-)
-from metadata.ingestion.api.steps import InvalidSourceException
-from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.database.lineage_source import LineageSource
-from metadata.utils.logger import ingestion_logger
-
-logger = ingestion_logger()
+from metadata.ingestion.source.database.mysql.queries import MYSQL_SQL_STATEMENT
+from metadata.ingestion.source.database.mysql.query_parser import MysqlQueryParserSource
 
 
-class MysqlLineageSource(LineageSource):
+class MysqlLineageSource(MysqlQueryParserSource, LineageSource):
+    sql_stmt = MYSQL_SQL_STATEMENT
+    filters = """
+        AND (
+            lower(argument) LIKE '%%create%%table%%select%%'
+            OR lower(argument) LIKE '%%insert%%into%%select%%'
+            OR lower(argument) LIKE '%%update%%'
+            OR lower(argument) LIKE '%%merge%%'
+        )
     """
-    Mysql lineage source implements view lineage
-    """
-
-    @classmethod
-    def create(
-        cls, config_dict, metadata: OpenMetadata, pipeline_name: Optional[str] = None
-    ):
-        """Create class instance"""
-        config: WorkflowSource = WorkflowSource.model_validate(config_dict)
-        connection: MysqlConnection = config.serviceConnection.root.config
-        if not isinstance(connection, MysqlConnection):
-            raise InvalidSourceException(
-                f"Expected MysqlConnection, but got {connection}"
-            )
-        return cls(config, metadata)

ingestion/src/metadata/ingestion/source/database/mysql/queries.py

@@ -0,0 +1,43 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""
+SQL Queries used during ingestion
+"""
+import textwrap
+
+MYSQL_SQL_STATEMENT = textwrap.dedent(
+    """
+SELECT 
+	NULL `database_name`,
+	argument `query_text`,
+	event_time `start_time`,
+    NULL `end_time`,
+	NULL `duration`,
+	NULL `schema_name`,
+    NULL `query_type`,
+    NULL `user_name`,
+    NULL `aborted`
+FROM mysql.general_log
+WHERE command_type = 'Query' 
+    AND event_time between '{start_time}' and '{end_time}'
+    AND argument NOT LIKE '/* {{"app": "OpenMetadata", %%}} */%%'
+    AND argument NOT LIKE '/* {{"app": "dbt", %%}} */%%'
+    {filters}
+ORDER BY event_time desc
+LIMIT {result_limit};
+"""
+)
+
+MYSQL_TEST_GET_QUERIES = textwrap.dedent(
+    """
+SELECT `argument` from mysql.general_log limit 1;
+"""
+)

ingestion/src/metadata/ingestion/source/database/mysql/query_parser.py

@@ -0,0 +1,46 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""
+Mysql query parser module
+"""
+from abc import ABC
+from typing import Optional
+
+from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
+    MysqlConnection,
+)
+from metadata.generated.schema.metadataIngestion.workflow import (
+    Source as WorkflowSource,
+)
+from metadata.ingestion.api.steps import InvalidSourceException
+from metadata.ingestion.ometa.ometa_api import OpenMetadata
+from metadata.ingestion.source.database.query_parser_source import QueryParserSource
+
+
+class MysqlQueryParserSource(QueryParserSource, ABC):
+    """
+    Mysql base for Usage and Lineage
+    """
+
+    filters: str
+
+    @classmethod
+    def create(
+        cls, config_dict, metadata: OpenMetadata, pipeline_name: Optional[str] = None
+    ):
+        """Create class instance"""
+        config: WorkflowSource = WorkflowSource.model_validate(config_dict)
+        connection: MysqlConnection = config.serviceConnection.root.config
+        if not isinstance(connection, MysqlConnection):
+            raise InvalidSourceException(
+                f"Expected MysqlConnection, but got {connection}"
+            )
+        return cls(config, metadata)

ingestion/src/metadata/ingestion/source/database/mysql/service_spec.py

@@ -1,4 +1,10 @@
+from metadata.ingestion.source.database.mysql.lineage import MysqlLineageSource
 from metadata.ingestion.source.database.mysql.metadata import MysqlSource
+from metadata.ingestion.source.database.mysql.usage import MysqlUsageSource
 from metadata.utils.service_spec.default import DefaultDatabaseSpec
 
-ServiceSpec = DefaultDatabaseSpec(metadata_source_class=MysqlSource)
+ServiceSpec = DefaultDatabaseSpec(
+    metadata_source_class=MysqlSource,
+    lineage_source_class=MysqlLineageSource,
+    usage_source_class=MysqlUsageSource,
+)

ingestion/src/metadata/ingestion/source/database/mysql/usage.py

@@ -0,0 +1,24 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""
+MYSQL usage module
+"""
+from metadata.ingestion.source.database.mysql.queries import MYSQL_SQL_STATEMENT
+from metadata.ingestion.source.database.mysql.query_parser import MysqlQueryParserSource
+from metadata.ingestion.source.database.usage_source import UsageSource
+
+
+class MysqlUsageSource(MysqlQueryParserSource, UsageSource):
+    sql_stmt = MYSQL_SQL_STATEMENT
+    filters = ""
+
+    def format_query(self, query: bytes) -> str:
+        return query.decode(errors="ignore").replace("\\n", "\n")

ingestion/tests/integration/automations/test_connection_automation.py

@@ -81,7 +81,7 @@ def test_connection_workflow(metadata, mysql_container):
     )
 
     assert final_workflow.status == WorkflowStatus.Successful
-    assert len(final_workflow.response.steps) == 4
+    assert len(final_workflow.response.steps) == 5
     assert final_workflow.response.status.value == StatusType.Successful.value
 
     metadata.delete(

ingestion/tests/integration/ometa/test_ometa_connection_definition_api.py

@@ -45,5 +45,5 @@ def test_get_connection_def():
     res: TestConnectionDefinition = metadata.get_by_name(
         entity=TestConnectionDefinition, fqn="Mysql.testConnectionDefinition"
     )
-    assert len(res.steps) == 4
+    assert len(res.steps) == 5
     assert res.name.root == "Mysql"

openmetadata-docs/content/v1.5.x/connectors/database/mysql/index.md

@@ -44,6 +44,18 @@ GRANT SELECT ON world.* TO '<username>';
 GRANT SELECT ON world.hello TO '<username>';
 ```
 
+### Lineage & Usage 
+To extract lineage & usage you need to enable the query logging in mysql and the user used in the connection needs to have select access to the `mysql.general_log`.
+
+```sql
+-- Enable Logging 
+SET GLOBAL general_log='ON';
+set GLOBAL log_output='table';
+
+-- Grant SELECT on log table
+GRANT SELECT ON mysql.general_log TO '<username>'@'<host>';
+```
+
 ### Profiler & Data Quality
 Executing the profiler workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. More information on the profiler workflow setup can be found [here](/how-to-guides/data-quality-observability/profiler/workflow) and data quality tests [here](/how-to-guides/data-quality-observability/quality).
 

openmetadata-docs/content/v1.5.x/connectors/database/mysql/yaml.md

@@ -36,6 +36,41 @@ To run the MySQL ingestion, you will need to install:
 pip3 install "openmetadata-ingestion[mysql]"
 ```
 
+### Metadata
+
+Note that We support MySQL (version 8.0.0 or greater) and the user should have access to the `INFORMATION_SCHEMA` table.  By default a user can see only the rows in the `INFORMATION_SCHEMA` that correspond to objects for which the user has the proper access privileges.
+
+```SQL
+-- Create user. If <hostName> is omitted, defaults to '%'
+-- More details https://dev.mysql.com/doc/refman/8.0/en/create-user.html
+CREATE USER '<username>'[@'<hostName>'] IDENTIFIED BY '<password>';
+
+-- Grant select on a database
+GRANT SELECT ON world.* TO '<username>';
+
+-- Grant select on a database
+GRANT SELECT ON world.* TO '<username>';
+
+-- Grant select on a specific object
+GRANT SELECT ON world.hello TO '<username>';
+```
+
+### Lineage & Usage 
+To extract lineage & usage you need to enable the query logging in mysql and the user used in the connection needs to have select access to the `mysql.general_log`.
+
+```sql
+-- Enable Logging 
+SET GLOBAL general_log='ON';
+set GLOBAL log_output='table';
+
+-- Grant SELECT on log table
+GRANT SELECT ON mysql.general_log TO '<username>'@'<host>';
+```
+
+### Profiler & Data Quality
+Executing the profiler workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. More information on the profiler workflow setup can be found [here](/how-to-guides/data-quality-observability/profiler/workflow) and data quality tests [here](/how-to-guides/data-quality-observability/quality).
+
+
 ## Metadata Ingestion
 
 All connectors are defined as JSON Schemas.

openmetadata-docs/content/v1.6.x-SNAPSHOT/connectors/database/mysql/index.md

@@ -44,6 +44,18 @@ GRANT SELECT ON world.* TO '<username>';
 GRANT SELECT ON world.hello TO '<username>';
 ```
 
+### Lineage & Usage 
+To extract lineage & usage you need to enable the query logging in mysql and the user used in the connection needs to have select access to the `mysql.general_log`.
+
+```sql
+-- Enable Logging 
+SET GLOBAL general_log='ON';
+set GLOBAL log_output='table';
+
+-- Grant SELECT on log table
+GRANT SELECT ON mysql.general_log TO '<username>'@'<host>';
+```
+
 ### Profiler & Data Quality
 Executing the profiler workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. More information on the profiler workflow setup can be found [here](/how-to-guides/data-quality-observability/profiler/workflow) and data quality tests [here](/how-to-guides/data-quality-observability/quality).
 

openmetadata-docs/content/v1.6.x-SNAPSHOT/connectors/database/mysql/yaml.md

@@ -36,6 +36,42 @@ To run the MySQL ingestion, you will need to install:
 pip3 install "openmetadata-ingestion[mysql]"
 ```
 
+### Metadata
+
+Note that We support MySQL (version 8.0.0 or greater) and the user should have access to the `INFORMATION_SCHEMA` table.  By default a user can see only the rows in the `INFORMATION_SCHEMA` that correspond to objects for which the user has the proper access privileges.
+
+```SQL
+-- Create user. If <hostName> is omitted, defaults to '%'
+-- More details https://dev.mysql.com/doc/refman/8.0/en/create-user.html
+CREATE USER '<username>'[@'<hostName>'] IDENTIFIED BY '<password>';
+
+-- Grant select on a database
+GRANT SELECT ON world.* TO '<username>';
+
+-- Grant select on a database
+GRANT SELECT ON world.* TO '<username>';
+
+-- Grant select on a specific object
+GRANT SELECT ON world.hello TO '<username>';
+```
+
+### Lineage & Usage 
+To extract lineage & usage you need to enable the query logging in mysql and the user used in the connection needs to have select access to the `mysql.general_log`.
+
+```sql
+-- Enable Logging 
+SET GLOBAL general_log='ON';
+set GLOBAL log_output='table';
+
+-- Grant SELECT on log table
+GRANT SELECT ON mysql.general_log TO '<username>'@'<host>';
+```
+
+### Profiler & Data Quality
+Executing the profiler workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. More information on the profiler workflow setup can be found [here](/how-to-guides/data-quality-observability/profiler/workflow) and data quality tests [here](/how-to-guides/data-quality-observability/quality).
+
+
+
 ## Metadata Ingestion
 
 All connectors are defined as JSON Schemas.

openmetadata-service/src/main/resources/json/data/testConnections/database/mysql.json

@@ -27,7 +27,12 @@
       "description": "From a given schema, list the views belonging to that schema. If no schema is specified, we'll list the tables of a random schema.",
       "errorMessage": "Failed to fetch views, please validate if the user has enough privilege to fetch views.",
       "mandatory": false
+    },
+    {
+      "name": "GetQueries",
+      "description": "Check if we can access the mysql.general_log table to get query logs, These queries are analyzed in the usage & lineage workflow.",
+      "errorMessage": "Failed to fetch queries, please validate if user can access the mysql.general_log table to get query logs.",
+      "mandatory": false
     }
   ]
-}
-
+}

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/mysqlConnection.json

@@ -107,8 +107,11 @@
         "title": "Supports Data Diff Extraction.",
         "$ref": "../connectionBasicType.json#/definitions/supportsDataDiff"
     },
-    "supportsViewLineageExtraction": {
-      "$ref": "../connectionBasicType.json#/definitions/supportsViewLineageExtraction"
+    "supportsUsageExtraction": {
+      "$ref": "../connectionBasicType.json#/definitions/supportsUsageExtraction"
+    },
+    "supportsLineageExtraction": {
+      "$ref": "../connectionBasicType.json#/definitions/supportsLineageExtraction"
     }
   },
   "additionalProperties": false,

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Mysql.md

@@ -24,6 +24,18 @@ $$note
 OpenMetadata supports MySQL version `8.0.0` and up. 
 $$
 
+### Lineage & Usage 
+To extract lineage & usage you need to enable the query logging in mysql and the user used in the connection needs to have select access to the `mysql.general_log`.
+
+```sql
+-- Enable Logging 
+SET GLOBAL general_log='ON';
+set GLOBAL log_output='table';
+
+-- Grant SELECT on log table
+GRANT SELECT ON mysql.general_log TO '<username>'@'<host>';
+```
+
 ### Profiler & Data Quality
 Executing the profiler Workflow or data quality tests, will require the user to have `SELECT` permission on the tables/schemas where the profiler/tests will be executed. The user should also be allowed to view information in `tables` for all objects in the database. More information on the profiler workflow setup can be found [here](https://docs.open-metadata.org/how-to-guides/data-quality-observability/profiler/workflow) and data quality tests [here](https://docs.open-metadata.org/connectors/ingestion/workflows/data-quality).