diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 97c41e7ee2e1..9495b3a82401 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -1633,6 +1633,10 @@ def add_input_dataset(self, name, dataset=None, dataset_id=None):
 
     def add_output_dataset(self, name, dataset):
         joda = JobToOutputDatasetAssociation(name, dataset)
+        if dataset.dataset.job is None:
+            # Only set job if dataset doesn't already have associated job.
+            # database operation tools that make copies should not modify the job here.
+            dataset.dataset.job = self
         add_object_to_object_session(self, joda)
         self.output_datasets.append(joda)
 
@@ -1996,24 +2000,8 @@ def update_output_states(self, supports_skip_locked):
             SET
                 state = :state,
                 update_time = :update_time
-            WHERE id IN (
-                SELECT hda.dataset_id FROM history_dataset_association hda
-                INNER JOIN job_to_output_dataset jtod
-                ON jtod.dataset_id = hda.id AND jtod.job_id = :job_id
-            );
-        """
-            ),
-            text(
-                """
-            UPDATE dataset
-            SET
-                state = :state,
-                update_time = :update_time
-            WHERE id IN (
-                SELECT ldda.dataset_id FROM library_dataset_dataset_association ldda
-                INNER JOIN job_to_output_library_dataset jtold
-                ON jtold.ldda_id = ldda.id AND jtold.job_id = :job_id
-            );
+            WHERE
+                dataset.job_id = :job_id
         """
             ),
             text(
@@ -2022,11 +2010,10 @@ def update_output_states(self, supports_skip_locked):
             SET
                 info = :info,
                 update_time = :update_time
-            WHERE id IN (
-                SELECT jtod.dataset_id
-                FROM job_to_output_dataset jtod
-                WHERE jtod.job_id = :job_id
-            );
+            FROM dataset
+            WHERE
+                history_dataset_association.dataset_id = dataset.id
+                AND dataset.job_id = :job_id;
         """
             ),
             text(
@@ -2035,11 +2022,10 @@ def update_output_states(self, supports_skip_locked):
             SET
                 info = :info,
                 update_time = :update_time
-            WHERE id IN (
-                SELECT jtold.ldda_id
-                FROM job_to_output_library_dataset jtold
-                WHERE jtold.job_id = :job_id
-            );
+            FROM dataset
+            WHERE
+                library_dataset_dataset_association.dataset_id = dataset.id
+                AND dataset.job_id = :job_id;
         """
             ),
         ]
diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py
index fa7aec93de4b..0b20c0e3c56d 100644
--- a/lib/galaxy/tools/execute.py
+++ b/lib/galaxy/tools/execute.py
@@ -175,11 +175,6 @@ def execute_single_job(execution_slice, completed_job, skip=False):
             history = execution_slice.history or history
             jobs_executed += 1
 
-    if job_datasets:
-        for job, datasets in job_datasets.items():
-            for dataset_instance in datasets:
-                dataset_instance.dataset.job = job
-
     if execution_slice:
         history.add_pending_items()
     # Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets
diff --git a/test/unit/app/managers/test_JobConnectionsManager.py b/test/unit/app/managers/test_JobConnectionsManager.py
index a9d482d2b916..5ec06f388fd7 100644
--- a/test/unit/app/managers/test_JobConnectionsManager.py
+++ b/test/unit/app/managers/test_JobConnectionsManager.py
@@ -25,10 +25,10 @@ def job_connections_manager(sa_session) -> JobConnectionsManager:
 
 # =============================================================================
 def setup_connected_dataset(sa_session: galaxy_scoped_session):
-    center_hda = HistoryDatasetAssociation(sa_session=sa_session)
-    input_hda = HistoryDatasetAssociation(sa_session=sa_session)
+    center_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    input_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     input_hdca = HistoryDatasetCollectionAssociation()
-    output_hda = HistoryDatasetAssociation(sa_session=sa_session)
+    output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     output_hdca = HistoryDatasetCollectionAssociation()
     input_job = Job()
     output_job = Job()
@@ -56,10 +56,10 @@ def setup_connected_dataset(sa_session: galaxy_scoped_session):
 
 def setup_connected_dataset_collection(sa_session: galaxy_scoped_session):
     center_hdca = HistoryDatasetCollectionAssociation()
-    input_hda1 = HistoryDatasetAssociation(sa_session=sa_session)
-    input_hda2 = HistoryDatasetAssociation(sa_session=sa_session)
+    input_hda1 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
+    input_hda2 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     input_hdca = HistoryDatasetCollectionAssociation()
-    output_hda = HistoryDatasetAssociation(sa_session=sa_session)
+    output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
     output_hdca = HistoryDatasetCollectionAssociation()
     input_job = Job()
     output_job = Job()