[DOP-31721] Log applied SQL transformation

Author: dolfinus

syncmaster/worker/handlers/db/base.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import logging
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any, ClassVar
 
@@ -20,6 +21,9 @@
     from syncmaster.dto.transfers import DBTransferDTO
 
 
+logger = logging.getLogger(__name__)
+
+
 class DBHandler(Handler):
     connection: BaseDBConnection
     transfer_dto: DBTransferDTO
@@ -62,6 +66,7 @@ def read(self) -> DataFrame:
 
         sql_query = self._get_sql_query()
         if sql_query:
+            logger.info("Applying SQL transformation:\n%s", sql_query)
             df.createOrReplaceTempView("source")
             df = self.connection.spark.sql(sql_query)
 

syncmaster/worker/handlers/file/local_df.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -18,6 +19,9 @@
     from pyspark.sql import DataFrame
 
 
+logger = logging.getLogger(__name__)
+
+
 class LocalDFFileHandler(FileHandler):
     local_df_connection: SparkLocalFS
 
@@ -60,6 +64,7 @@ def read(self) -> DataFrame:
 
         sql_query = self._get_sql_query()
         if sql_query:
+            logger.info("Applying SQL transformation:\n%s", sql_query)
             df.createOrReplaceTempView("source")
             df = self.df_connection.spark.sql(sql_query)
 

syncmaster/worker/handlers/file/remote_df.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -14,6 +15,9 @@
     from pyspark.sql import DataFrame
 
 
+logger = logging.getLogger(__name__)
+
+
 class RemoteDFFileHandler(FileHandler):
     def read(self) -> DataFrame:
         from pyspark.sql.types import StructType  # noqa: PLC0415
@@ -37,6 +41,7 @@ def read(self) -> DataFrame:
 
         sql_query = self._get_sql_query()
         if sql_query:
+            logger.info("Applying SQL transformation:\n%s", sql_query)
             df.createOrReplaceTempView("source")
             df = self.df_connection.spark.sql(sql_query)
 

syncmaster/worker/handlers/file/s3.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING
 
 from onetl.file import FileDFReader
@@ -16,6 +17,9 @@
     from syncmaster.dto.connections import S3ConnectionDTO
 
 
+logger = logging.getLogger(__name__)
+
+
 @support_hooks
 class S3Handler(RemoteDFFileHandler):
     connection_dto: S3ConnectionDTO
@@ -73,6 +77,7 @@ def read(self) -> DataFrame:
 
         sql_query = self._get_sql_query()
         if sql_query:
+            logger.info("Applying SQL transformation:\n%s", sql_query)
             df.createOrReplaceTempView("source")
             df = self.df_connection.spark.sql(sql_query)