perf(records): add in-process caches and benchmark tooling

cds_migrator_kit/rdm/records/transform/entities/parent.py

@@ -19,6 +19,10 @@
 
 EMAIL_PATTERN = re.compile(r"[^@]+@[^@]+\.[^@]+")
 
+# Cache submitter email → user_id across records.  Stable for one migration
+# run since users are created before the records stream starts.
+_submitter_id_by_email: dict = {}
+
 
 class RecordParent:
     """The parent record for one migrated CDS record.
@@ -77,9 +81,12 @@ def _build_access(self):
         email = self.dojson_entry.pop("submitter", None)
         if not email:
             owner = "system"
+        elif email in _submitter_id_by_email:
+            owner = _submitter_id_by_email[email]
         else:
             try:
                 user = User.query.filter_by(email=email).one()
+                _submitter_id_by_email[email] = user.id
                 owner = user.id
             except NoResultFound:
                 raise UnexpectedValue(

cds_migrator_kit/rdm/records/transform/entities/request.py

@@ -12,6 +12,12 @@
 
 from cds_migrator_kit.errors import ManualImportRequired, RecordFlaggedCuration
 
+# Cache reviewer lookup results across records.  Maps the raw reviewer string
+# (email or "Family, Given" name) to the resolved user_id (int) on success, or
+# to the RecordFlaggedCuration exception instance when no account was found.
+# Stable for one migration run: user accounts don't change during records stream.
+_reviewer_id_cache: dict = {}
+
 
 class RecordRequest:
     """A community-inclusion request for one migrated CDS record.
@@ -67,14 +73,24 @@ def _resolve_reviewers(self, reviewer_names):
         """
         resolved = []
         for reviewer_name in reviewer_names:
-            try:
-                user = self._find_reviewer(reviewer_name)
-                reviewer_entry = {"user": str(user.id)}
-            except RecordFlaggedCuration as exc:
+            cached = _reviewer_id_cache.get(reviewer_name)
+            if cached is None and reviewer_name not in _reviewer_id_cache:
+                try:
+                    user = self._find_reviewer(reviewer_name)
+                    _reviewer_id_cache[reviewer_name] = user.id
+                    cached = user.id
+                except RecordFlaggedCuration as exc:
+                    _reviewer_id_cache[reviewer_name] = exc
+                    cached = exc
+
+            if isinstance(cached, RecordFlaggedCuration):
                 self.migration_logger.add_information(
-                    self.recid, {"message": exc.message, "value": exc.value}
+                    self.recid, {"message": cached.message, "value": cached.value}
                 )
                 reviewer_entry = {"user": "-1"}
+            else:
+                reviewer_entry = {"user": str(cached)}
+
             if reviewer_entry not in resolved:
                 resolved.append(reviewer_entry)
         return resolved

cds_migrator_kit/rdm/records/transform/mappers/contributors.py

@@ -20,14 +20,32 @@
 from cds_migrator_kit.rdm.migration_config import VOCABULARIES_NAMES_SCHEMES
 from cds_migrator_kit.rdm.records.transform.mappers.base import FieldMapper
 
+# Sentinel distinguishing "not yet cached" from "cached as None (not found)".
+_MISSING = object()
+
+# Per-process caches. Plain-dict row representations avoid SQLAlchemy object
+# expiry issues that occur across UoW commits between records.
+_affiliation_ror_cache: dict = (
+    {}
+)  # ror_normalized → bool (exists in AffiliationsMetadata)
+_affiliation_legacy_cache: dict = (
+    {}
+)  # legacy_name → {"curated","exact","not_exact"} or None
+_person_id_to_user_id: dict = {}  # cern person_id str → user_id int or None
+
 
 def match_affiliation(affiliation_name, ctx):
     """Match an affiliation against `CDSMigrationAffiliationMapping` db table."""
     dojson_entry = ctx.dojson_entry
     if is_ror(affiliation_name):
         ror = normalize_ror(affiliation_name)
-        name = AffiliationsMetadata.query.filter_by(pid=ror).one_or_none()
-        if name is None:
+        exists = _affiliation_ror_cache.get(ror, _MISSING)
+        if exists is _MISSING:
+            exists = (
+                AffiliationsMetadata.query.filter_by(pid=ror).one_or_none() is not None
+            )
+            _affiliation_ror_cache[ror] = exists
+        if not exists:
             raise ManualImportRequired(
                 message="Affiliation {ror} does not exist in the AffiliationMetadata table".format(
                     ror=ror
@@ -41,20 +59,35 @@ def match_affiliation(affiliation_name, ctx):
                 subfield=None,
             )
         return {"id": normalize_ror(affiliation_name)}
-    # Step 1: search in the affiliation mapping (ROR organizations)
-    match = ctx.affiliations_mapping.query.filter_by(
-        legacy_affiliation_input=affiliation_name
-    ).one_or_none()
-    if match:
+
+    # Legacy name lookup — cache the row fields as a plain dict to avoid
+    # SQLAlchemy object expiry that occurs across UoW commits.
+    cached = _affiliation_legacy_cache.get(affiliation_name, _MISSING)
+    if cached is _MISSING:
+        match = ctx.affiliations_mapping.query.filter_by(
+            legacy_affiliation_input=affiliation_name
+        ).one_or_none()
+        cached = (
+            {
+                "curated": match.curated_affiliation,
+                "exact": match.ror_exact_match,
+                "not_exact": match.ror_not_exact_match,
+            }
+            if match is not None
+            else None
+        )
+        _affiliation_legacy_cache[affiliation_name] = cached
+
+    if cached is not None:
         # Step 1: check if there is a curated input
-        if match.curated_affiliation:
-            return match.curated_affiliation
+        if cached["curated"]:
+            return cached["curated"]
         # Step 2: check if there is an exact match
-        if match.ror_exact_match:
-            return {"id": normalize_ror(match.ror_exact_match)}
+        if cached["exact"]:
+            return {"id": normalize_ror(cached["exact"])}
         # Step 3: check if there is not exact match
-        if match.ror_not_exact_match:
-            _affiliation_ror_id = normalize_ror(match.ror_not_exact_match)
+        if cached["not_exact"]:
+            _affiliation_ror_id = normalize_ror(cached["not_exact"])
             raise RecordFlaggedCuration(
                 subfield="u",
                 value={"id": _affiliation_ror_id},
@@ -120,9 +153,12 @@ def _lookup_person_id(creator):
         {},
     ).get("identifier")
     if person_id:
-        ui = UserIdentity.query.filter_by(id=person_id).one_or_none()
-        if ui:
-            user_id = ui.user.id
+        user_id = _person_id_to_user_id.get(person_id, _MISSING)
+        if user_id is _MISSING:
+            ui = UserIdentity.query.filter_by(id=person_id).one_or_none()
+            user_id = ui.user.id if ui else None
+            _person_id_to_user_id[person_id] = user_id
+        if user_id is not None:
             names = NamesMetadata.query.filter_by(internal_id=str(user_id)).all()
             name = next(
                 (name for name in names if "unlisted" not in name.json.get("tags", [])),

cds_migrator_kit/reports/log.py

@@ -88,6 +88,8 @@ def __init__(
         ]
         self.log_writer = csv.DictWriter(self.error_file, fieldnames=columns)
         self._temp_state_cache = {}
+        self._flush_interval = 100
+        self._rows_since_flush = 0
 
     def start_log(self):
         """Initialize logging file descriptors."""
@@ -116,6 +118,7 @@ def read_log(self):
 
     def finalise(self):
         """Finalise logging files."""
+        self.error_file.flush()
         self.error_file.close()
 
     def add_log(self, exc, record=None, key=None, value=None):
@@ -145,7 +148,10 @@ def add_log(self, exc, record=None, key=None, value=None):
         }
         self.log_writer.writerow(error_format)
         logger_migrator.error(exc)
-        self.error_file.flush()
+        self._rows_since_flush += 1
+        if self._rows_since_flush >= self._flush_interval:
+            self.error_file.flush()
+            self._rows_since_flush = 0
 
     def add_information(self, recid, state):
         """Save a temporary success state for recid.
@@ -163,7 +169,10 @@ def finalise_record(self, recid):
         """Log recid as success."""
         _state = self._temp_state_cache.pop(recid, {})
         self.log_writer.writerow({"recid": recid, "clean": True, **_state})
-        self.error_file.flush()
+        self._rows_since_flush += 1
+        if self._rows_since_flush >= self._flush_interval:
+            self.error_file.flush()
+            self._rows_since_flush = 0
 
 
 class RecordStateLogger:

cds_migrator_kit/runner/runner.py

@@ -7,6 +7,8 @@
 
 """InvenioRDM migration streams runner."""
 
+import logging
+import time
 from pathlib import Path
 
 import yaml
@@ -120,15 +122,21 @@ def __init__(
 
     def run(self):
         """Run ETL streams."""
+        perf_logger = logging.getLogger("migrator-perf")
 
         self.migration_logger.start_log()
         self.record_state_logger.start_log()
         for stream in self.streams:
+            t0 = time.perf_counter()
             try:
                 stream.run(cleanup=True)
             except Exception as e:
                 self.migration_logger.add_log(e)
                 raise e
             finally:
+                elapsed = time.perf_counter() - t0
+                perf_logger.info(
+                    f"Stream '{stream.name}' finished in {elapsed:.1f}s"
+                )
                 self.migration_logger.finalise()
                 self.record_state_logger.finalise()