update: only update the missing hashes with --to-remote

[isidentical]

dvc update --to-remote now only updates the missing cache files instead of syncing all over, as a pre-requisite this patch also merges transfer() logic with stage() and introduces the stage(upload=...) option for transferring to the remote cache.

Fixes #4992, part of #5768

[efiop]

@isidentical Cool optimization! 🔥 We've talked about getting rid of transfer() in favor of stage() before, I think it needs to be done as a pre-requisite for this optimization, as it seems transfer itself gets more complicated and actually update case is almost indistinguishable from normal stage(). And because of this additional similarity, I think it is worth doing that in this PR.

[isidentical]

Benchmark results: master, this branch. The import-url --to-remote performance is reduced from 3.50 min to 2.10 min, and the update --to-remote performance is reduced from 9~ min to 3~ min.

[efiop]

Seems like you could executor.map, or am I missing something?

[isidentical]

We need to manually unpack some attributes of entry (entry.path_info, entry.fs etc) + add **kwargs to the calls. In theory we can do something like executor.map(lambda entry: odb.add(...), ...), what would be the advantage with map though? It would just block the progress bar if there is some big file in the middle, which might not be the best UI.

[efiop]

Seemed nicer and shorter. E.g. see how we wrap the worker in tqdm in dvc/objects/stage.py, where we use it in map, so it doesn't block the pbar.

[efiop]

@isidentical Btw, very interesting failures on windows. Looks like we have the same objects (e.g. two files with the same content) saved in parallel, which results in that move exception. We need to simply ignore it (somewhere in objects/db/base.py likely), as it is an expected behavior for when someone (e.g. another user) is saving stuff in parallel to the same cache dir.

EDIT: that will fix #4992

[efiop]

Looks great! 🔥 As we've discussed above, there are a few things like using verify in odb.add and putting stage()&save() logic somewhere in one method, but let's deal with those on top.