refactor: further logging details

Author: stevenhsd

src/dve/core_engine/backends/base/rules.py

@@ -413,8 +413,12 @@ def apply_sync_filters(
                     if not success:
                         return messages, False
 
+                self.logger.info(f"Filter {rule.reporting.code} found {len(temp_messages)} issues")
+
             if filter_column_names:
-                self.logger.info("Filtering records where validation is record level")
+                self.logger.info(
+                    f"Filtering records from entity {entity_name} for error code {rule.reporting.code}" # pylint: disable=line-too-long
+                )
                 success_condition = " AND ".join(
                     [f"({c_name} IS NOT NULL AND {c_name})" for c_name in filter_column_names]
                 )
@@ -476,7 +480,7 @@ def apply_rules(self, entities: Entities, rule_metadata: RuleMetadata) -> Messag
             rules_and_locals = rule_metadata
 
         messages: Messages = []
-        
+
         self.logger.info("Applying pre-sync steps")
         for rule, local_variables in rules_and_locals:
             for step in rule.pre_sync_steps:
@@ -505,7 +509,7 @@ def apply_rules(self, entities: Entities, rule_metadata: RuleMetadata) -> Messag
             return messages
 
         self.logger.info("Applying post-sync steps")
-        
+
         for rule, local_variables in rules_and_locals:
             for step in rule.post_sync_steps:
                 if rule_metadata.templating_strategy == "runtime":

src/dve/core_engine/backends/implementations/duckdb/contract.py

@@ -102,7 +102,6 @@ def apply_data_contract(
         self, entities: DuckDBEntities, contract_metadata: DataContractMetadata
     ) -> tuple[DuckDBEntities, Messages, StageSuccessful]:
         """Apply the data contract to the duckdb relations"""
-        self.logger.info("Applying data contracts")
         all_messages: Messages = []
 
         successful = True
@@ -131,6 +130,9 @@ def apply_data_contract(
             coerce_inferred_numpy_array_to_list(relation.df()).apply(
                 application_helper, axis=1
             )  # pandas uses eager evaluation so potential memory issue here?
+            self.logger.info(
+                f"Data contract found {len(application_helper.errors)} issues in {entity_name}"
+            )
             all_messages.extend(application_helper.errors)
 
             casting_statements = [

src/dve/core_engine/backends/implementations/spark/contract.py

@@ -113,7 +113,10 @@ def apply_data_contract(
                 # .persist(storageLevel=StorageLevel.MEMORY_AND_DISK)
             )
             messages = validated.flatMap(lambda row: row[1]).filter(bool)
+            messages.cache()
+            self.logger.info(f"Data contract found {messages.count()} issues in {entity_name}")
             all_messages.extend(messages.collect())
+            messages.unpersist()
 
             try:
                 record_df = record_df.select(

src/dve/pipeline/duckdb_pipeline.py

@@ -21,6 +21,7 @@ class DDBDVEPipeline(BaseDVEPipeline):
     """
     Modified Pipeline class for running a DVE Pipeline with Spark
     """
+
     # pylint: disable=R0913
     def __init__(
         self,

src/dve/pipeline/pipeline.py

@@ -231,6 +231,7 @@ def audit_received_file_step(
         self, pool: ThreadPoolExecutor, submitted_files: Iterable[tuple[FileURI, InfoURI]]
     ) -> tuple[list[SubmissionInfo], list[SubmissionInfo]]:
         """Set files as being received and mark them for file transformation"""
+        self._logger.info("Starting audit received file service")
         audit_received_futures: list[tuple[str, FileURI, Future]] = []
         for submission_file in submitted_files:
             data_uri, metadata_uri = submission_file
@@ -292,7 +293,7 @@ def file_transformation(
         """Transform a file from its original format into a 'stringified' parquet file"""
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
-
+        self._logger.info(f"Applying file transformation to {submission_info.submission_id}")
         errors: list[FeedbackMessage] = []
         submission_status: SubmissionStatus = SubmissionStatus()
         submission_file_uri: URI = fh.joinuri(
@@ -327,6 +328,7 @@ def file_transformation_step(
         list[tuple[SubmissionInfo, SubmissionStatus]], list[tuple[SubmissionInfo, SubmissionStatus]]
     ]:
         """Step to transform files from their original format into parquet files"""
+        self._logger.info("Starting file transformation service")
         file_transform_futures: list[tuple[SubmissionInfo, Future]] = []
 
         for submission_info in submissions_to_process:
@@ -398,6 +400,7 @@ def apply_data_contract(
         self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
     ) -> tuple[SubmissionInfo, SubmissionStatus]:
         """Method for applying the data contract given a submission_info"""
+        self._logger.info(f"Applying data contract to {submission_info.submission_id}")
         if not submission_status:
             submission_status = self.get_submission_status(
                 "contract", submission_info.submission_id
@@ -451,6 +454,7 @@ def data_contract_step(
         list[tuple[SubmissionInfo, SubmissionStatus]], list[tuple[SubmissionInfo, SubmissionStatus]]
     ]:
         """Step to validate the types of an untyped (stringly typed) parquet file"""
+        self._logger.info("Starting data contract service")
         processed_files: list[tuple[SubmissionInfo, SubmissionStatus]] = []
         failed_processing: list[tuple[SubmissionInfo, SubmissionStatus]] = []
         dc_futures: list[tuple[SubmissionInfo, SubmissionStatus, Future]] = []
@@ -518,6 +522,7 @@ def apply_business_rules(
         """Apply the business rules to a given submission, the submission may have failed at the
         data_contract step so this should be passed in as a bool
         """
+        self._logger.info(f"Applying business rules to {submission_info.submission_id}")
         if not submission_status:
             submission_status = self.get_submission_status(
                 "business_rules", submission_info.submission_id
@@ -607,6 +612,7 @@ def business_rule_step(
         list[tuple[SubmissionInfo, SubmissionStatus]],
     ]:
         """Step to apply business rules (Step impl) to a typed parquet file"""
+        self._logger.info("Starting business rules service")
         future_files: list[tuple[SubmissionInfo, SubmissionStatus, Future]] = []
 
         for submission_info, submission_status in files:
@@ -748,7 +754,7 @@ def error_report(
         SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]
     ]:
         """Creates the error reports given a submission info and submission status"""
-        self._logger.info("Generating error report")
+        self._logger.info(f"Generating error report for {submission_info.submission_id}")
         if not submission_status:
             submission_status = self.get_submission_status(
                 "error_report", submission_info.submission_id
@@ -816,6 +822,7 @@ def error_report_step(
         """Step to produce error reports
         takes processed files and files that failed file transformation
         """
+        self._logger.info("Starting error reports service")
         futures: list[tuple[SubmissionInfo, SubmissionStatus, Future]] = []
         reports: list[
             tuple[SubmissionInfo, SubmissionStatus, Union[None, SubmissionStatisticsRecord], URI]

src/dve/pipeline/spark_pipeline.py

@@ -23,6 +23,7 @@ class SparkDVEPipeline(BaseDVEPipeline):
     """
     Polymorphed Pipeline class for running a DVE Pipeline with Spark
     """
+
     # pylint: disable=R0913
     def __init__(
         self,