[DOP-19903] - add integration MySQL tests & CI

Author: maxim-lixakov

.env.docker

@@ -91,6 +91,14 @@ TEST_MSSQL_USER=syncmaster
 TEST_MSSQL_PASSWORD=7ellowEl7akey
 TEST_MSSQL_DB=syncmaster
 
+TEST_MYSQL_HOST_FOR_CONFTEST=test-mysql
+TEST_MYSQL_PORT_FOR_CONFTEST=3306
+TEST_MYSQL_HOST_FOR_WORKER=test-mysql
+TEST_MYSQL_PORT_FOR_WORKER=3306
+TEST_MYSQL_USER=syncmaster
+TEST_MYSQL_PASSWORD=ohbuz9Eochaj9saibooK3thooGa5aesh
+TEST_MYSQL_DB=syncmaster
+
 TEST_HIVE_CLUSTER=test-hive
 TEST_HIVE_USER=syncmaster
 TEST_HIVE_PASSWORD=changeme

.env.local

@@ -78,6 +78,14 @@ export TEST_MSSQL_USER=syncmaster
 export TEST_MSSQL_PASSWORD=7ellowEl7akey
 export TEST_MSSQL_DB=syncmaster
 
+export TEST_MYSQL_HOST_FOR_CONFTEST=localhost
+export TEST_MYSQL_PORT_FOR_CONFTEST=3306
+export TEST_MYSQL_HOST_FOR_WORKER=test-mysql
+export TEST_MYSQL_PORT_FOR_WORKER=3306
+export TEST_MYSQL_USER=syncmaster
+export TEST_MYSQL_PASSWORD=ohbuz9Eochaj9saibooK3thooGa5aesh
+export TEST_MYSQL_DB=syncmaster
+
 export TEST_HIVE_CLUSTER=test-hive
 export TEST_HIVE_USER=syncmaster
 export TEST_HIVE_PASSWORD=changeme

.github/workflows/mysql-tests.yml

@@ -0,0 +1,76 @@
+name: MySQL Tests
+on:
+  workflow_call:
+
+env:
+  DEFAULT_PYTHON: '3.12'
+
+jobs:
+  tests:
+    name: Run MySQL tests
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Cache jars
+        uses: actions/cache@v4
+        with:
+          path: ./cached_jars
+          key: ${{ runner.os }}-python-${{ env.DEFAULT_PYTHON }}-test-mysql
+          restore-keys: |
+            ${{ runner.os }}-python-${{ env.DEFAULT_PYTHON }}-test-mysql
+            ${{ runner.os }}-python-
+      - name: Build Worker Image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          tags: mtsrus/syncmaster-worker:${{ github.sha }}
+          target: test
+          file: docker/Dockerfile.worker
+          load: true
+          cache-from: mtsrus/syncmaster-worker:develop
+
+      - name: Docker compose up
+        run: |
+          docker compose -f docker-compose.test.yml --profile all down -v --remove-orphans
+          docker compose -f docker-compose.test.yml --profile mysql up -d --wait --wait-timeout 200
+        env:
+          WORKER_IMAGE_TAG: ${{ github.sha }}
+
+      - name: Run MySQL Tests
+        run: |
+          docker compose -f ./docker-compose.test.yml --profile mysql exec -T worker coverage run -m pytest -vvv -s -m "worker and mysql"
+      - name: Dump worker logs on failure
+        if: failure()
+        uses: jwalton/gh-docker-logs@v2
+        with:
+          images: mtsrus/syncmaster-worker
+          dest: ./logs
+
+      # This is important, as coverage is exported after receiving SIGTERM
+      - name: Shutdown
+        if: always()
+        run: |
+          docker compose -f docker-compose.test.yml --profile all down -v --remove-orphans
+      - name: Upload worker logs
+        uses: actions/upload-artifact@v4
+        if: failure()
+        with:
+          name: worker-logs-mysql
+          path: logs/*
+
+      - name: Upload coverage results
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-mysql
+          path: reports/*
+          # https://github.com/actions/upload-artifact/issues/602
+          include-hidden-files: true

.github/workflows/tests.yml

@@ -24,6 +24,10 @@ jobs:
     name: Clickhouse tests
     uses: ./.github/workflows/clickhouse-tests.yml
 
+  mysql_tests:
+    name: MySQL tests
+    uses: ./.github/workflows/mysql-tests.yml
+
   mssql_tests:
     name: MSSQL tests
     uses: ./.github/workflows/mssql-tests.yml
@@ -52,7 +56,7 @@ jobs:
     name: Tests done
     runs-on: ubuntu-latest
 
-    needs: [oracle_tests, clickhouse_tests, mssql_tests, hive_tests, hdfs_tests, s3_tests, unit_tests]
+    needs: [oracle_tests, clickhouse_tests, mssql_tests, mysql_tests, hive_tests, hdfs_tests, s3_tests, unit_tests]
     steps:
       - name: Checkout code
         uses: actions/checkout@v4

Makefile

@@ -93,6 +93,10 @@ test-integration-clickhouse: test-db ##@Test    Run integration tests for Clickh
 	docker compose -f docker-compose.test.yml --profile clickhouse up -d --wait $(DOCKER_COMPOSE_ARGS)
 	${POETRY} run pytest ./tests/test_integration -m clickhouse $(PYTEST_ARGS)
 
+test-integration-mysql: test-db ##@Test         Run integration tests for MySQL
+	docker compose -f docker-compose.test.yml --profile mysql up -d --wait $(DOCKER_COMPOSE_ARGS)
+	${POETRY} run pytest ./tests/test_integration -m mysql $(PYTEST_ARGS)
+
 test-integration-mssql: test-db ##@Test         Run integration tests for MSSQL
 	docker compose -f docker-compose.test.yml --profile mssql up -d --wait $(DOCKER_COMPOSE_ARGS)
 	${POETRY} run pytest ./tests/test_integration -m mssql $(PYTEST_ARGS)

docker-compose.test.yml

@@ -96,7 +96,7 @@ services:
         condition: service_healthy
       rabbitmq:
         condition: service_healthy
-    profiles: [worker, scheduler, s3, oracle, hdfs, hive, clickhouse, mssql, all]
+    profiles: [worker, scheduler, s3, oracle, hdfs, hive, clickhouse, mysql, mssql, all]
 
   test-postgres:
     image: postgres
@@ -110,7 +110,7 @@ services:
       interval: 30s
       timeout: 5s
       retries: 3
-    profiles: [s3, oracle, clickhouse, mssql, hdfs, hive, all]
+    profiles: [s3, oracle, clickhouse, mysql, mssql, hdfs, hive, all]
 
   test-s3:
     image: bitnami/minio:latest
@@ -167,6 +167,20 @@ services:
     platform: linux/amd64
     profiles: [mssql, all]
 
+  test-mysql:
+    image: mysql
+    restart: unless-stopped
+    environment:
+      MYSQL_ROOT_PASSWORD: ohbuz9Eochaj9saibooK3thooGa5aesh
+      MYSQL_DATABASE: syncmaster
+      MYSQL_USER: syncmaster
+      MYSQL_PASSWORD: ohbuz9Eochaj9saibooK3thooGa5aesh
+    ports:
+      - 3306:3306
+    platform: linux/amd64
+    profiles: [mysql, all]
+
+
   metastore-hive:
     image: postgres
     restart: unless-stopped

syncmaster/dto/connections.py

@@ -42,6 +42,17 @@ class MSSQLConnectionDTO(ConnectionDTO):
     type: ClassVar[str] = "mssql"
 
 
+@dataclass
+class MySQLConnectionDTO(ConnectionDTO):
+    host: str
+    port: int
+    user: str
+    password: str
+    database_name: str
+    additional_params: dict
+    type: ClassVar[str] = "mysql"
+
+
 @dataclass
 class OracleConnectionDTO(ConnectionDTO):
     host: str

syncmaster/dto/transfers.py

@@ -61,6 +61,11 @@ class MSSQLTransferDTO(DBTransferDTO):
     type: ClassVar[str] = "mssql"
 
 
+@dataclass
+class MySQLTransferDTO(DBTransferDTO):
+    type: ClassVar[str] = "mysql"
+
+
 @dataclass
 class HiveTransferDTO(DBTransferDTO):
     type: ClassVar[str] = "hive"

syncmaster/worker/controller.py

@@ -9,6 +9,7 @@
     HDFSConnectionDTO,
     HiveConnectionDTO,
     MSSQLConnectionDTO,
+    MySQLConnectionDTO,
     OracleConnectionDTO,
     PostgresConnectionDTO,
     S3ConnectionDTO,
@@ -18,6 +19,7 @@
     HDFSTransferDTO,
     HiveTransferDTO,
     MSSQLTransferDTO,
+    MySQLTransferDTO,
     OracleTransferDTO,
     PostgresTransferDTO,
     S3TransferDTO,
@@ -27,6 +29,7 @@
 from syncmaster.worker.handlers.db.clickhouse import ClickhouseHandler
 from syncmaster.worker.handlers.db.hive import HiveHandler
 from syncmaster.worker.handlers.db.mssql import MSSQLHandler
+from syncmaster.worker.handlers.db.mysql import MySQLHandler
 from syncmaster.worker.handlers.db.oracle import OracleHandler
 from syncmaster.worker.handlers.db.postgres import PostgresHandler
 from syncmaster.worker.handlers.file.hdfs import HDFSHandler
@@ -57,6 +60,11 @@
         MSSQLConnectionDTO,
         MSSQLTransferDTO,
     ),
+    "mysql": (
+        MySQLHandler,
+        MySQLConnectionDTO,
+        MySQLTransferDTO,
+    ),
     "postgres": (
         PostgresHandler,
         PostgresConnectionDTO,

syncmaster/worker/handlers/db/mysql.py

@@ -0,0 +1,37 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from onetl.connection import MySQL
+
+from syncmaster.dto.connections import MySQLConnectionDTO
+from syncmaster.dto.transfers import MySQLTransferDTO
+from syncmaster.worker.handlers.db.base import DBHandler
+
+if TYPE_CHECKING:
+    from pyspark.sql import SparkSession
+    from pyspark.sql.dataframe import DataFrame
+
+
+class MySQLHandler(DBHandler):
+    connection: MySQL
+    connection_dto: MySQLConnectionDTO
+    transfer_dto: MySQLTransferDTO
+
+    def connect(self, spark: SparkSession):
+        self.connection = MySQL(
+            host=self.connection_dto.host,
+            port=self.connection_dto.port,
+            user=self.connection_dto.user,
+            password=self.connection_dto.password,
+            database=self.connection_dto.database_name,
+            spark=spark,
+        ).check()
+
+    def normalize_column_names(self, df: DataFrame) -> DataFrame:
+        for column_name in df.columns:
+            df = df.withColumnRenamed(column_name, column_name.lower())
+        return df