[DOP-28871] Fix handling Spark unknown job name

Author: dolfinus

data_rentgen/dto/job.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+from copy import copy
 from dataclasses import dataclass, field
 
 from data_rentgen.dto.job_type import JobTypeDTO
@@ -22,14 +23,17 @@ def unique_key(self) -> tuple:
 
     def merge(self, new: JobDTO) -> JobDTO:
         self.id = new.id or self.id
-        self.location.merge(new.location)
-
-        if self.name == "unknown" and new.name != "unknown":
-            # Workaround for https://github.com/OpenLineage/OpenLineage/issues/3846
-            self.name = new.name
+        self.location = self.location.merge(new.location)
 
         if new.type and self.type:
-            self.type.merge(new.type)
+            self.type = self.type.merge(new.type)
         else:
             self.type = new.type or self.type
+
+        if self.name == "unknown" and new.name != "unknown":
+            # Workaround for https://github.com/OpenLineage/OpenLineage/issues/3846
+            result = copy(self)
+            result.name = new.name
+            return result
+
         return self

data_rentgen/dto/operation.py

@@ -58,9 +58,9 @@ def unique_key(self) -> tuple:
         return (self.id,)
 
     def merge(self, new: OperationDTO) -> OperationDTO:
-        self.run.merge(new.run)
+        self.run = self.run.merge(new.run)
         if self.sql_query and new.sql_query:
-            self.sql_query.merge(new.sql_query)
+            self.sql_query = self.sql_query.merge(new.sql_query)
         else:
             self.sql_query = new.sql_query or self.sql_query
 

data_rentgen/dto/output.py

@@ -63,11 +63,11 @@ def generate_id(self) -> UUID:
         return generate_incremental_uuid(self.created_at, ".".join(id_components))
 
     def merge(self, new: OutputDTO) -> OutputDTO:
-        self.operation.merge(new.operation)
-        self.dataset.merge(new.dataset)
+        self.operation = self.operation.merge(new.operation)
+        self.dataset = self.dataset.merge(new.dataset)
 
         if self.schema and new.schema:
-            self.schema.merge(new.schema)
+            self.schema = self.schema.merge(new.schema)
         else:
             self.schema = new.schema or self.schema
 

data_rentgen/dto/run.py

@@ -53,14 +53,14 @@ def unique_key(self) -> tuple:
         return (self.id,)
 
     def merge(self, new: RunDTO) -> RunDTO:
-        self.job.merge(new.job)
+        self.job = self.job.merge(new.job)
         if new.parent_run and self.parent_run:
-            self.parent_run.merge(new.parent_run)
+            self.parent_run = self.parent_run.merge(new.parent_run)
         else:
             self.parent_run = new.parent_run or self.parent_run
 
         if new.user and self.user:
-            self.user.merge(new.user)
+            self.user = self.user.merge(new.user)
         else:
             self.user = new.user or self.user
 

tests/test_consumer/test_extractors/fixtures/spark_dto.py

@@ -41,6 +41,17 @@ def extracted_spark_app_job(
     )
 
 
+@pytest.fixture
+def extracted_spark_unknown_job(
+    extracted_spark_location: LocationDTO,
+) -> JobDTO:
+    return JobDTO(
+        name="unknown",
+        location=extracted_spark_location,
+        type=JobTypeDTO(type="SPARK_APPLICATION"),
+    )
+
+
 @pytest.fixture
 def extracted_spark_app_run(
     extracted_spark_app_job: JobDTO,

tests/test_consumer/test_extractors/test_extractors_batch_spark.py

@@ -100,6 +100,7 @@ def test_extractors_extract_batch_spark_openlineage_emitted_unknown_name(
     spark_operation_run_event_stop: OpenLineageRunEvent,
     extracted_spark_location: LocationDTO,
     extracted_spark_app_job: JobDTO,
+    extracted_spark_unknown_job: JobDTO,
     extracted_user: UserDTO,
     extracted_spark_app_run: RunDTO,
     extracted_spark_operation: OperationDTO,
@@ -119,7 +120,7 @@ def test_extractors_extract_batch_spark_openlineage_emitted_unknown_name(
         extracted_spark_location,
     ]
 
-    assert extracted.jobs() == [extracted_spark_app_job]
+    assert extracted.jobs() == [extracted_spark_app_job, extracted_spark_unknown_job]
     assert extracted.users() == [extracted_user]
     assert extracted.runs() == [extracted_spark_app_run]
     assert extracted.operations() == [extracted_spark_operation]