diff --git a/src/dve/core_engine/backends/base/contract.py b/src/dve/core_engine/backends/base/contract.py index 948ff77..003f0b4 100644 --- a/src/dve/core_engine/backends/base/contract.py +++ b/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: diff --git a/src/dve/core_engine/backends/metadata/contract.py b/src/dve/core_engine/backends/metadata/contract.py index e3eb0c0..2e7565d 100644 --- a/src/dve/core_engine/backends/metadata/contract.py +++ b/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.""" diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py index 10a452d..4469b98 100644 --- a/src/dve/core_engine/configuration/v1/__init__.py +++ b/src/dve/core_engine/configuration/v1/__init__.py @@ -192,7 +192,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) @@ -387,7 +387,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): diff --git a/src/dve/metadata_parser/models.py b/src/dve/metadata_parser/models.py index 49e2386..0b7d535 100644 --- a/src/dve/metadata_parser/models.py +++ b/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) diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index ee5a6bc..255e1eb 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -585,7 +585,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 @@ -621,15 +620,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( @@ -726,9 +721,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 diff --git a/tests/test_pipeline/test_foundry_ddb_pipeline.py b/tests/test_pipeline/test_foundry_ddb_pipeline.py index 9b7b60d..f84073a 100644 --- a/tests/test_pipeline/test_foundry_ddb_pipeline.py +++ b/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 diff --git a/tests/test_pipeline/test_spark_pipeline.py b/tests/test_pipeline/test_spark_pipeline.py index dd28e26..ee9b085 100644 --- a/tests/test_pipeline/test_spark_pipeline.py +++ b/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()