Do not fetch /preupload if already done in upload-large-folder (#3100)

* Batch preupload calls in upload-large-folder

* Revert "Batch preupload calls in upload-large-folder"

This reverts commit 73754c4.

* upload mode was not fetched

* type hints

* make quality again

* let fix #3016 as well

Author: Wauplin

src/huggingface_hub/_local_folder.py

@@ -149,6 +149,7 @@ class LocalUploadFileMetadata:
     should_ignore: Optional[bool] = None
     sha256: Optional[str] = None
     upload_mode: Optional[str] = None
+    remote_oid: Optional[str] = None
     is_uploaded: bool = False
     is_committed: bool = False
 
@@ -172,6 +173,9 @@ def save(self, paths: LocalUploadFilePaths) -> None:
 
                 if self.upload_mode is not None:
                     f.write(self.upload_mode)
+
+                if self.remote_oid is not None:
+                    f.write(self.remote_oid)
                 f.write("\n")
 
                 f.write(str(int(self.is_uploaded)) + "\n")
@@ -346,6 +350,9 @@ def read_upload_metadata(local_dir: Path, filename: str) -> LocalUploadFileMetad
                     if upload_mode not in (None, "regular", "lfs"):
                         raise ValueError(f"Invalid upload mode in metadata {paths.path_in_repo}: {upload_mode}")
 
+                    _remote_oid = f.readline().strip()
+                    remote_oid = None if _remote_oid == "" else _remote_oid
+
                     is_uploaded = bool(int(f.readline().strip()))
                     is_committed = bool(int(f.readline().strip()))
 
@@ -355,6 +362,7 @@ def read_upload_metadata(local_dir: Path, filename: str) -> LocalUploadFileMetad
                         should_ignore=should_ignore,
                         sha256=sha256,
                         upload_mode=upload_mode,
+                        remote_oid=remote_oid,
                         is_uploaded=is_uploaded,
                         is_committed=is_committed,
                     )

src/huggingface_hub/_upload_large_folder.py

@@ -42,8 +42,7 @@
 logger = logging.getLogger(__name__)
 
 WAITING_TIME_IF_NO_TASKS = 10  # seconds
-MAX_NB_REGULAR_FILES_PER_COMMIT = 75
-MAX_NB_LFS_FILES_PER_COMMIT = 150
+MAX_NB_FILES_FETCH_UPLOAD_MODE = 100
 COMMIT_SIZE_SCALE: List[int] = [20, 50, 75, 100, 125, 200, 250, 400, 600, 1000]
 
 
@@ -404,19 +403,19 @@ def _determine_next_job(status: LargeUploadStatus) -> Optional[Tuple[WorkerJob,
         ):
             status.nb_workers_commit += 1
             logger.debug("Job: commit (more than 5 minutes since last commit attempt)")
-            return (WorkerJob.COMMIT, _get_items_to_commit(status.queue_commit))
+            return (WorkerJob.COMMIT, _get_n(status.queue_commit, status.target_chunk()))
 
         # 2. Commit if at least 100 files are ready to commit
         elif status.nb_workers_commit == 0 and status.queue_commit.qsize() >= 150:
             status.nb_workers_commit += 1
             logger.debug("Job: commit (>100 files ready)")
-            return (WorkerJob.COMMIT, _get_items_to_commit(status.queue_commit))
+            return (WorkerJob.COMMIT, _get_n(status.queue_commit, status.target_chunk()))
 
-        # 3. Get upload mode if at least 10 files
-        elif status.queue_get_upload_mode.qsize() >= 10:
+        # 3. Get upload mode if at least 100 files
+        elif status.queue_get_upload_mode.qsize() >= MAX_NB_FILES_FETCH_UPLOAD_MODE:
             status.nb_workers_get_upload_mode += 1
-            logger.debug("Job: get upload mode (>10 files ready)")
-            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, status.target_chunk()))
+            logger.debug(f"Job: get upload mode (>{MAX_NB_FILES_FETCH_UPLOAD_MODE} files ready)")
+            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, MAX_NB_FILES_FETCH_UPLOAD_MODE))
 
         # 4. Preupload LFS file if at least 1 file and no worker is preuploading LFS
         elif status.queue_preupload_lfs.qsize() > 0 and status.nb_workers_preupload_lfs == 0:
@@ -434,7 +433,7 @@ def _determine_next_job(status: LargeUploadStatus) -> Optional[Tuple[WorkerJob,
         elif status.queue_get_upload_mode.qsize() > 0 and status.nb_workers_get_upload_mode == 0:
             status.nb_workers_get_upload_mode += 1
             logger.debug("Job: get upload mode (no other worker getting upload mode)")
-            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, status.target_chunk()))
+            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, MAX_NB_FILES_FETCH_UPLOAD_MODE))
 
         # 7. Preupload LFS file if at least 1 file
         #    Skip if hf_transfer is enabled and there is already a worker preuploading LFS
@@ -455,7 +454,7 @@ def _determine_next_job(status: LargeUploadStatus) -> Optional[Tuple[WorkerJob,
         elif status.queue_get_upload_mode.qsize() > 0:
             status.nb_workers_get_upload_mode += 1
             logger.debug("Job: get upload mode")
-            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, status.target_chunk()))
+            return (WorkerJob.GET_UPLOAD_MODE, _get_n(status.queue_get_upload_mode, MAX_NB_FILES_FETCH_UPLOAD_MODE))
 
         # 10. Commit if at least 1 file and 1 min since last commit attempt
         elif (
@@ -466,7 +465,7 @@ def _determine_next_job(status: LargeUploadStatus) -> Optional[Tuple[WorkerJob,
         ):
             status.nb_workers_commit += 1
             logger.debug("Job: commit (1 min since last commit attempt)")
-            return (WorkerJob.COMMIT, _get_items_to_commit(status.queue_commit))
+            return (WorkerJob.COMMIT, _get_n(status.queue_commit, status.target_chunk()))
 
         # 11. Commit if at least 1 file all other queues are empty and all workers are waiting
         #     e.g. when it's the last commit
@@ -482,7 +481,7 @@ def _determine_next_job(status: LargeUploadStatus) -> Optional[Tuple[WorkerJob,
         ):
             status.nb_workers_commit += 1
             logger.debug("Job: commit")
-            return (WorkerJob.COMMIT, _get_items_to_commit(status.queue_commit))
+            return (WorkerJob.COMMIT, _get_n(status.queue_commit, status.target_chunk()))
 
         # 12. If all queues are empty, exit
         elif all(metadata.is_committed or metadata.should_ignore for _, metadata in status.items):
@@ -528,6 +527,7 @@ def _get_upload_mode(items: List[JOB_ITEM_T], api: "HfApi", repo_id: str, repo_t
         paths, metadata = item
         metadata.upload_mode = addition._upload_mode
         metadata.should_ignore = addition._should_ignore
+        metadata.remote_oid = addition._remote_oid
         metadata.save(paths)
 
 
@@ -580,6 +580,9 @@ def _build_hacky_operation(item: JOB_ITEM_T) -> HackyCommitOperationAdd:
     if metadata.sha256 is None:
         raise ValueError("sha256 must have been computed by now!")
     operation.upload_info = UploadInfo(sha256=bytes.fromhex(metadata.sha256), size=metadata.size, sample=sample)
+    operation._upload_mode = metadata.upload_mode  # type: ignore[assignment]
+    operation._should_ignore = metadata.should_ignore
+    operation._remote_oid = metadata.remote_oid
     return operation
 
 
@@ -596,30 +599,6 @@ def _get_n(queue: "queue.Queue[JOB_ITEM_T]", n: int) -> List[JOB_ITEM_T]:
     return [queue.get() for _ in range(min(queue.qsize(), n))]
 
 
-def _get_items_to_commit(queue: "queue.Queue[JOB_ITEM_T]") -> List[JOB_ITEM_T]:
-    """Special case for commit job: the number of items to commit depends on the type of files."""
-    # Can take at most 50 regular files and/or 100 LFS files in a single commit
-    items: List[JOB_ITEM_T] = []
-    nb_lfs, nb_regular = 0, 0
-    while True:
-        # If empty queue => commit everything
-        if queue.qsize() == 0:
-            return items
-
-        # If we have enough items => commit them
-        if nb_lfs >= MAX_NB_LFS_FILES_PER_COMMIT or nb_regular >= MAX_NB_REGULAR_FILES_PER_COMMIT:
-            return items
-
-        # Else, get a new item and increase counter
-        item = queue.get()
-        items.append(item)
-        _, metadata = item
-        if metadata.upload_mode == "lfs":
-            nb_lfs += 1
-        else:
-            nb_regular += 1
-
-
 def _print_overwrite(report: str) -> None:
     """Print a report, overwriting the previous lines.
 

src/huggingface_hub/hf_api.py

@@ -4421,20 +4421,23 @@ def preupload_lfs_files(
         new_additions = [addition for addition in additions if not addition._is_uploaded]
 
         # Check which new files are LFS
-        try:
-            _fetch_upload_modes(
-                additions=new_additions,
-                repo_type=repo_type,
-                repo_id=repo_id,
-                headers=headers,
-                revision=revision,
-                endpoint=self.endpoint,
-                create_pr=create_pr or False,
-                gitignore_content=gitignore_content,
-            )
-        except RepositoryNotFoundError as e:
-            e.append_to_message(_CREATE_COMMIT_NO_REPO_ERROR_MESSAGE)
-            raise
+        # For some items, we might have already fetched the upload mode (in case of upload_large_folder)
+        additions_no_upload_mode = [addition for addition in new_additions if addition._upload_mode is None]
+        if len(additions_no_upload_mode) > 0:
+            try:
+                _fetch_upload_modes(
+                    additions=additions_no_upload_mode,
+                    repo_type=repo_type,
+                    repo_id=repo_id,
+                    headers=headers,
+                    revision=revision,
+                    endpoint=self.endpoint,
+                    create_pr=create_pr or False,
+                    gitignore_content=gitignore_content,
+                )
+            except RepositoryNotFoundError as e:
+                e.append_to_message(_CREATE_COMMIT_NO_REPO_ERROR_MESSAGE)
+                raise
 
         # Filter out regular files
         new_lfs_additions = [addition for addition in new_additions if addition._upload_mode == "lfs"]