DM-31824: Use ResourcePath.mtransfer for Butler.transfer_from

python/lsst/daf/butler/cli/cmd/commands.py

@@ -638,6 +638,9 @@ def retrieve_artifacts(**kwargs: Any) -> None:
 @transfer_option()
 @register_dataset_types_option()
 @transfer_dimensions_option()
+@click.option(
+    "--dry-run/--no-dry-run", default=False, help="Enable dry run mode and do not transfer any datasets."
+)
 @options_file_option()
 def transfer_datasets(**kwargs: Any) -> None:
     """Transfer datasets from a source butler to a destination butler.

python/lsst/daf/butler/datastores/fileDatastore.py

@@ -2158,10 +2158,19 @@ def retrieveArtifacts(
                     target_uri = to_transfer[cleaned_source_uri]
                     artifact_map[target_uri].append(ref.id)
 
-        # In theory can now parallelize the transfer
+        # Parallelize the transfer. Re-raise as a single exception if
+        # a FileExistsError is encountered anywhere.
         log.debug("Number of artifacts to transfer to %s: %d", str(destination), len(to_transfer))
-        for source_uri, target_uri in to_transfer.items():
-            target_uri.transfer_from(source_uri, transfer=transfer, overwrite=overwrite)
+        try:
+            ResourcePath.mtransfer(
+                transfer,
+                ((from_uri, to_uri) for from_uri, to_uri in to_transfer.items()),
+                overwrite=overwrite,
+            )
+        except* FileExistsError as egroup:
+            raise FileExistsError(
+                "Some files already exist in destination directory and overwrite is False"
+            ) from egroup
 
         # Transfer the Zip files and unpack them.
         zipped_artifacts = unpack_zips(zips_to_transfer, requested_ids, destination, preserve_path, overwrite)
@@ -2748,6 +2757,9 @@ def transfer_from(
         # Record each time we have done a "direct" transfer.
         direct_transfers = []
 
+        # Keep track of all the file transfers that are required.
+        from_to: list[tuple[ResourcePath, ResourcePath]] = []
+
         # Now can transfer the artifacts
         log.verbose("Transferring artifacts")
         for ref in refs:
@@ -2796,21 +2808,28 @@ def transfer_from(
                             info = info.update(path=target_location.pathInStore.path)
 
                     # Need to transfer it to the new location.
-                    # Assume we should always overwrite. If the artifact
-                    # is there this might indicate that a previous transfer
-                    # was interrupted but was not able to be rolled back
-                    # completely (eg pre-emption) so follow Datastore default
-                    # and overwrite. Do not copy if we are in dry-run mode.
-                    if not dry_run:
-                        target_location.uri.transfer_from(
-                            source_location.uri,
-                            transfer=transfer,
-                            overwrite=True,
-                            transaction=self._transaction,
-                        )
+                    from_to.append((source_location.uri, target_location.uri))
 
                 artifacts.append((ref, info))
 
+        # Do the file transfers in bulk.
+        # Assume we should always overwrite. If the artifact
+        # is there this might indicate that a previous transfer
+        # was interrupted but was not able to be rolled back
+        # completely (eg pre-emption) so follow Datastore default
+        # and overwrite. Do not copy if we are in dry-run mode.
+        if dry_run:
+            log.info("Would be copying %d file artifacts", len(from_to))
+        else:
+            log.verbose("Copying %d file artifacts", len(from_to))
+            with time_this(log, msg="Transferring datasets into datastore", level=VERBOSE):
+                ResourcePath.mtransfer(
+                    transfer,
+                    from_to,
+                    overwrite=True,
+                    transaction=self._transaction,
+                )
+
         if direct_transfers:
             log.info(
                 "Transfer request for an outside-datastore artifact with absolute URI done %d time%s",

python/lsst/daf/butler/script/transferDatasets.py

@@ -49,6 +49,7 @@ def transferDatasets(
     transfer: str,
     register_dataset_types: bool,
     transfer_dimensions: bool = True,
+    dry_run: bool = False,
 ) -> int:
     """Transfer datasets from run in source to dest.
 
@@ -84,6 +85,9 @@ def transferDatasets(
         Indicate whether dimensions should be transferred along with
         datasets. It can be more efficient to disable this if it is known
         that all dimensions exist.
+    dry_run : `bool`, optional
+        If `True` no transfers are done but the number of transfers that
+        would be done is reported.
     """
     source_butler = Butler.from_config(source, writeable=False)
     dest_butler = Butler.from_config(dest, writeable=True)
@@ -112,5 +116,6 @@ def transferDatasets(
         transfer=transfer,
         register_dataset_types=register_dataset_types,
         transfer_dimensions=transfer_dimensions,
+        dry_run=dry_run,
     )
     return len(transferred)

requirements.txt

@@ -3,5 +3,5 @@
 
 lsst-sphgeom @ git+https://github.com/lsst/sphgeom@main
 lsst-utils @ git+https://github.com/lsst/utils@main
-lsst-resources[https,s3] @ git+https://github.com/lsst/resources@main
+lsst-resources[https,s3] @ git+https://github.com/lsst/resources@tickets/DM-31824
 lsst-daf-relation @ git+https://github.com/lsst/daf_relation@main