refactor: merging logging additions from release branch

Author: stevenhsd

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

@@ -378,13 +378,15 @@ def apply_sync_filters(
             logger=self.logger,
         ) as msg_writer:
             for entity_name, filter_rules in filters_by_entity.items():
+                self.logger.info(f"Applying filters to {entity_name}")
                 entity = entities[entity_name]
 
                 filter_column_names: list[str] = []
                 unmodified_entities = {entity_name: entity}
                 modified_entities = {entity_name: entity}
 
                 for rule in filter_rules:
+                    self.logger.info(f"Applying filter {rule.reporting.code}")
                     if rule.reporting.emit == "record_failure":
                         column_name = f"filter_{uuid4().hex}"
                         filter_column_names.append(column_name)
@@ -436,6 +438,9 @@ def apply_sync_filters(
                             return processing_errors_uri, False
                         if temp_messages:
                             msg_writer.write_queue.put(temp_messages)
+                        self.logger.info(
+                            f"Filter {rule.reporting.code} found {len(temp_messages)} issues"
+                            )
 
                     else:
                         temp_messages, success = self.evaluate(
@@ -461,8 +466,15 @@ def apply_sync_filters(
                             return processing_errors_uri, False
                         if temp_messages:
                             msg_writer.write_queue.put(temp_messages)
+                        
+                        self.logger.info(
+                            f"Filter {rule.reporting.code} found {len(temp_messages)} issues"
+                            )
 
                 if filter_column_names:
+                    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]
                     )
@@ -537,6 +549,7 @@ def apply_rules(
         altering the entities in-place.
 
         """
+        self.logger.info("Applying business rules")
         rules_and_locals: Iterable[tuple[Rule, TemplateVariables]]
         errors_uri = get_feedback_errors_uri(working_directory, "business_rules")
         if rule_metadata.templating_strategy == "upfront":
@@ -554,6 +567,7 @@ def apply_rules(
             rules_and_locals = rule_metadata
 
         pre_sync_messages: Messages = []
+        self.logger.info("Applying pre-sync steps")
         for rule, local_variables in rules_and_locals:
             for step in rule.pre_sync_steps:
                 if rule_metadata.templating_strategy == "runtime":
@@ -601,6 +615,8 @@ def apply_rules(
             return errors_uri, False
 
         post_sync_messages: 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

@@ -163,10 +163,16 @@ def apply_data_contract(
                 )
 
                 batches = pq.ParquetFile(entity_locations[entity_name]).iter_batches(10000)
+                msg_count = 0
                 with Pool(8) as pool:
                     for msgs in pool.imap_unordered(row_validator_helper, batches):
                         if msgs:
                             msg_writer.write_queue.put(msgs)
+                            msg_count += len(msgs)
+                
+                self.logger.info(
+                f"Data contract found {msg_count} issues in {entity_name}"
+            )
 
                 casting_statements = [
                     (

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

@@ -129,11 +129,14 @@ def apply_data_contract(
                 working_dir, "data_contract", key_fields, self.logger
             ) as msg_writer:
                 messages = validated.flatMap(lambda row: row[1]).filter(bool).toLocalIterator()
+                msg_count = 0
                 while True:
                     batch = list(islice(messages, 10000))
                     if not batch:
                         break
                     msg_writer.write_queue.put(batch)
+                    msg_count += len(batch)
+                self.logger.info(f"Data contract found {msg_count} issues in {entity_name}")
 
             try:
                 record_df = record_df.select(

src/dve/pipeline/pipeline.py

@@ -195,6 +195,7 @@ def write_file_to_parquet(
         errors = []
 
         for model_name, model in models.items():
+            self._logger.info(f"Transforming {model_name} to stringified parquet")
             reader: BaseFileReader = load_reader(dataset, model_name, ext)
             try:
                 if not entity_type:
@@ -235,6 +236,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
@@ -296,7 +298,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(
@@ -331,6 +333,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:
@@ -402,6 +405,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(
                 "data_contract", submission_info.submission_id
@@ -456,6 +460,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]] = []
@@ -523,6 +528,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
@@ -613,6 +619,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:
@@ -754,7 +761,7 @@ def error_report(
         SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]
     ]:
         """Creates the error reports given a submission info and submission status"""
-
+        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
@@ -763,6 +770,7 @@ def error_report(
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
 
+        self._logger.info("Reading error dataframes")
         errors_df, aggregates = self._get_error_dataframes(submission_info.submission_id)
 
         if not submission_status.number_of_records:
@@ -801,9 +809,11 @@ def error_report(
             "error_reports",
             f"{submission_info.file_name}_{submission_info.file_extension.strip('.')}.xlsx",
         )
+        self._logger.info("Writing error report")
         with fh.open_stream(report_uri, "wb") as stream:
             stream.write(er.ExcelFormat.convert_to_bytes(workbook))
 
+        self._logger.info("Publishing error aggregates")
         self._publish_error_aggregates(submission_info.submission_id, aggregates)
 
         return submission_info, submission_status, sub_stats, report_uri
@@ -819,6 +829,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]