Add optional bulk_remote_exists

Author: sjawhar

dvc/repo/data.py

@@ -456,15 +456,41 @@ def _onerror(entry, exc):
             n += 1
             cb.set_size(n)
 
-        for key, entry in entries.items():
-            k = (*key, "") if entry.meta and entry.meta.isdir else key
+        entries_list = list(entries.values())
+        if hasattr(storage_map, "bulk_remote_exists"):
             try:
-                if not storage_map.remote_exists(entry, refresh=remote_refresh):
-                    missing_entries.append(os.path.sep.join(k))
-            except StorageKeyError:
-                pass
-            finally:
-                cb.relative_update()
+                results = storage_map.bulk_remote_exists(
+                    entries_list, refresh=remote_refresh
+                )
+                for key, entry in entries.items():
+                    k = (*key, "") if entry.meta and entry.meta.isdir else key
+                    try:
+                        if not results.get(entry, True):
+                            missing_entries.append(os.path.sep.join(k))
+                    except StorageKeyError:
+                        pass
+                    finally:
+                        cb.relative_update()
+            except Exception:
+                for key, entry in entries.items():
+                    k = (*key, "") if entry.meta and entry.meta.isdir else key
+                    try:
+                        if not storage_map.remote_exists(entry, refresh=remote_refresh):
+                            missing_entries.append(os.path.sep.join(k))
+                    except StorageKeyError:
+                        pass
+                    finally:
+                        cb.relative_update()
+        else:
+            for key, entry in entries.items():
+                k = (*key, "") if entry.meta and entry.meta.isdir else key
+                try:
+                    if not storage_map.remote_exists(entry, refresh=remote_refresh):
+                        missing_entries.append(os.path.sep.join(k))
+                except StorageKeyError:
+                    pass
+                finally:
+                    cb.relative_update()
     data_index.onerror = orig_data_index_onerror
     return missing_entries