[DOP-22401] Small fixes and optimizations

data_rentgen/consumer/extractors/batch.py

@@ -86,10 +86,15 @@ def __repr__(self):
 
     @staticmethod
     def _add(context: dict[tuple, T], new_item: T) -> dict[tuple, T]:  # noqa: WPS602
-        if new_item.unique_key in context:
-            context[new_item.unique_key] = context[new_item.unique_key].merge(new_item)
+        key = new_item.unique_key
+        if key in context:
+            old_item = context[key]
+            if old_item is new_item:
+                return context
+
+            context[key] = old_item.merge(new_item)
         else:
-            context[new_item.unique_key] = new_item
+            context[key] = new_item
         return context
 
     def add_location(self, location: LocationDTO):

data_rentgen/consumer/extractors/dataset.py

@@ -58,8 +58,9 @@ def extract_dataset(dataset: OpenLineageDataset | OpenLineageSymlinkIdentifier)
 
 
 def extract_dataset_and_symlinks(dataset: OpenLineageDataset) -> tuple[DatasetDTO, list[DatasetSymlinkDTO]]:
+    dataset_dto = extract_dataset(dataset)
     if not dataset.facets.symlinks:
-        return extract_dataset(dataset), []
+        return dataset_dto, []
 
     table_symlinks = [
         identifier
@@ -78,38 +79,28 @@ def extract_dataset_and_symlinks(dataset: OpenLineageDataset) -> tuple[DatasetDT
                 "Dataset has more than one TABLE symlink. Only the first one will be used for replacement. Symlink name: %s",
                 table_symlinks[0].name,
             )
-        table_dataset = table_symlinks[0]
+        table_dataset_dto = extract_dataset(table_symlinks[0])
         return (
-            DatasetDTO(
-                name=table_dataset.name,
-                location=extract_dataset_location(table_dataset),
-                format=extract_dataset_format(table_dataset),
-            ),
+            table_dataset_dto,
             connect_dataset_with_symlinks(
-                extract_dataset(dataset),
-                extract_dataset(table_dataset),
+                dataset_dto,
+                table_dataset_dto,
                 OpenLineageSymlinkType.TABLE,
             ),
         )
 
     symlinks = []
-    for identifier in dataset.facets.symlinks.identifiers:
+    for symlink_identifier in dataset.facets.symlinks.identifiers:
+        symlink_dto = extract_dataset(symlink_identifier)
         symlinks.extend(
             connect_dataset_with_symlinks(
-                extract_dataset(dataset),
-                extract_dataset(identifier),
-                identifier.type,
+                dataset_dto,
+                symlink_dto,
+                symlink_identifier.type,
             ),
         )
 
-    return (
-        DatasetDTO(
-            name=dataset.name,
-            location=extract_dataset_location(dataset),
-            format=extract_dataset_format(dataset),
-        ),
-        symlinks,
-    )
+    return dataset_dto, symlinks
 
 
 def extract_dataset_location(dataset: OpenLineageDataset | OpenLineageSymlinkIdentifier) -> LocationDTO:

data_rentgen/db/models/custom_user_properties.py

@@ -45,6 +45,6 @@ class CustomUserProperties(Base):
 
     during: Mapped[TSTZRANGE] = mapped_column(
         TSTZRANGE,
-        server_defaults=text("tstzrange(now(), NULL, '[)')"),
+        server_default=text("tstzrange(now(), NULL, '[)')"),
         nullable=False,
     )