PR #736 no need to split up db_updates in _process_threadworker_updates

is not necessary and eliminates `job_db.read()` which is not part of JobDatabaseInterface

Author: soxofaan

openeo/extra/job_management/__init__.py

@@ -709,17 +709,9 @@ def _process_threadworker_updates(
         # Build DataFrame of updates indexed by df_idx
         df_updates = pd.DataFrame(updates).set_index("df_idx", drop=True)
 
-        # Determine which rows to upsert
-        existing_indices = set(df_updates.index).intersection(job_db.read().index)
-        if existing_indices:
-            df_upsert = df_updates.loc[sorted(existing_indices)]
-            job_db.persist(df_upsert)
-            stats["job_db persist"] = stats.get("job_db persist", 0) + 1
-
-        # Any df_idx not in original index are errors
-        missing = set(df_updates.index) - existing_indices
-        if missing:
-            _log.error(f"Skipping non-existing dataframe indices: {sorted(missing)}")
+        job_db.persist(df_updates)
+        stats["job_db persist"] = stats.get("job_db persist", 0) + 1
+
 
     def on_job_done(self, job: BatchJob, row):
         """

tests/extra/job_management/test_job_management.py

@@ -786,9 +786,6 @@ def test_process_threadworker_updates(self, tmp_path, caplog):
         assert stats.get("queued", 0) == 2
         assert stats["job_db persist"] == 1
 
-        # Assert error log for invalid index
-        assert any("Skipping non-existing dataframe indices" in msg for msg in caplog.messages)
-
     def test_no_results_leaves_db_and_stats_untouched(self, tmp_path, caplog):
         pool = _JobManagerWorkerThreadPool(max_workers=2)
         stats = collections.defaultdict(int)