refactor: remove original entities (#164)

redundant in the grand scheme of things and was increasing the I/O overhead

Author: georgeRobertson

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

@@ -443,10 +443,6 @@ def apply(
                 ],
             )
 
-        if contract_metadata.cache_originals:
-            for entity_name in list(entities):
-                entities[f"Original{entity_name}"] = entities[entity_name]
-
         return entities, feedback_errors_uri, successful, processing_errors_uri
 
     def read_parquet(self, path: URI, **kwargs) -> EntityType:

src/dve/core_engine/backends/metadata/contract.py

@@ -35,7 +35,7 @@ class DataContractMetadata(BaseModel, frozen=True, arbitrary_types_allowed=True)
     reporting_fields: dict[EntityName, ReportingFields]
     """The per-entity reporting fields."""
     cache_originals: bool = False
-    """Whether to cache the original entities after loading."""
+    """WARNING - Depreciated functionality. Whether to cache the original entities after loading."""
     _schemas: dict[EntityName, type[BaseModel]] = PrivateAttr(default_factory=dict)
     """The pydantic models of the schmas."""
 

src/dve/core_engine/configuration/v1/__init__.py

@@ -207,7 +207,7 @@ class V1DataContractConfig(BaseModel):
     """Configuration for the data contract component of the dataset."""
 
     cache_originals: bool = False
-    """Whether to cache the original entities after loading."""
+    """WARNING - Depreciated functionality. Whether to cache the original entities after loading."""
     error_details: Optional[URI] = None
     """Optional URI containing custom data contract error codes and messages"""
     types: dict[TypeName, TypeOrDef] = Field(default_factory=dict)
@@ -402,7 +402,6 @@ def get_contract_metadata(self) -> DataContractMetadata:
             reader_metadata=reader_metadata,
             validators=validators,
             reporting_fields=reporting_fields,
-            cache_originals=self.contract.cache_originals,
         )
 
     def load_error_message_info(self, uri):

src/dve/metadata_parser/models.py

@@ -391,6 +391,7 @@ class DatasetSpecification(BaseModel):
     """Configuration options for a dataset."""
 
     cache_originals: bool = False
+    """WARNING - Depreciated functionality."""
     types: dict[TypeName, FieldSpecification] = Field(default_factory=dict)
     """Predefined types to be used within schema/dataset definitions."""
     schemas: dict[EntityName, EntitySpecification] = Field(default_factory=dict)

src/dve/pipeline/pipeline.py

@@ -624,7 +624,6 @@ def apply_business_rules(  # pylint: disable=R0914,R0915
             entities[file_name] = self.step_implementations.add_record_index(  # type: ignore
                 self.step_implementations.read_parquet(parquet_uri)  # type: ignore
             )
-            entities[f"Original{file_name}"] = self.step_implementations.read_parquet(parquet_uri)  # type: ignore
 
         sub_info_entity = (
             self._audit_tables._submission_info.conv_to_entity(  # pylint: disable=protected-access
@@ -660,15 +659,11 @@ def apply_business_rules(  # pylint: disable=R0914,R0915
         for entity_name, entity in entity_manager.entities.items():
             # Note BI filtering done within the apply_rules
             self._logger.info(f"applying data contract filter to {entity_name}.")
-            if not entity_name.startswith("Original"):
-                filtered_entity = self._step_implementations.filter_data_contract_record_rejections(
-                    working_directory,
-                    entity,
-                    entity_name,
-                )
-            else:
-                self._logger.info(f"Skipping {entity_name}. Marked original.")
-                filtered_entity = entity
+            filtered_entity = self._step_implementations.filter_data_contract_record_rejections(
+                working_directory,
+                entity,
+                entity_name,
+            )
             projected = self._step_implementations.write_parquet(  # type: ignore
                 filtered_entity,
                 fh.joinuri(
@@ -769,9 +764,14 @@ def apply_business_rules(  # pylint: disable=R0914,R0915
         )
 
         submission_status.number_of_records = self.get_entity_count(
-            entity=entity_manager.entities[f"""Original{rules.global_variables.get(
-                                              'entity',
-                                              submission_info.dataset_id)}"""]
+            entity=self.step_implementations.read_parquet(  # type: ignore
+                fh.joinuri(
+                    self.processed_files_path,
+                    submission_info.submission_id,
+                    "data_contract",
+                    rules.global_variables.get('entity', submission_info.dataset_id)
+                )
+            )
         )
         submission_status.number_of_records_rejected = (
             submission_status.number_of_records

tests/test_pipeline/test_foundry_ddb_pipeline.py

@@ -102,7 +102,7 @@ def test_foundry_runner_validation_success(movies_test_files, temp_ddb_conn):
         )
         output_loc, report_uri, audit_files = dve_pipeline.run_pipeline(sub_info)
         assert fh.get_resource_exists(report_uri)
-        assert len(list(fh.iter_prefix(output_loc))) == 2
+        assert len(list(fh.iter_prefix(output_loc))) == 1
         assert len(list(fh.iter_prefix(audit_files))) == 3
 
 def test_foundry_runner_error(planet_test_files, temp_ddb_conn):
@@ -197,7 +197,7 @@ def test_foundry_runner_with_submitted_files_path(movies_test_files, temp_ddb_co
 
         assert Path(processing_folder, sub_id, sub_info.file_name_with_ext).exists()
         assert fh.get_resource_exists(report_uri)
-        assert len(list(fh.iter_prefix(output_loc))) == 2
+        assert len(list(fh.iter_prefix(output_loc))) == 1
         assert len(list(fh.iter_prefix(audit_files))) == 3
 
 

tests/test_pipeline/test_spark_pipeline.py

@@ -274,12 +274,6 @@ def test_apply_business_rules_success(
     assert largest_satellites_entity_path.exists()
     assert spark.read.parquet(str(largest_satellites_entity_path)).count() == 1
 
-    og_planets_entity_path = Path(
-        Path(processed_file_path), sub_info.submission_id, "business_rules", "Originalplanets"
-    )
-    assert og_planets_entity_path.exists()
-    assert spark.read.parquet(str(og_planets_entity_path)).count() == 1
-
 
 def test_apply_business_rules_with_data_errors(  # pylint: disable=redefined-outer-name
     spark: SparkSession,
@@ -317,10 +311,6 @@ def test_apply_business_rules_with_data_errors(  # pylint: disable=redefined-out
     assert largest_satellites_entity_path.exists()
     assert spark.read.parquet(str(largest_satellites_entity_path)).count() == 1
 
-    og_planets_entity_path = br_path / "Originalplanets"
-    assert og_planets_entity_path.exists()
-    assert spark.read.parquet(str(og_planets_entity_path)).count() == 1
-
     errors_path = Path(br_path.parent, "errors", "business_rules_errors.jsonl")
     assert errors_path.exists()