hdfs: copy: don't read everything into memory (#5277)

Regression from #4973 and #4747

dvc/tree/hdfs.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import re
+import shutil
 import subprocess
 from collections import deque
 from contextlib import closing, contextmanager
@@ -162,7 +163,7 @@ def copy(self, from_info, to_info, **_kwargs):
                     with closing(
                         hdfs.open_output_stream(tmp_info.path)
                     ) as tmp_fobj:
-                        tmp_fobj.write(from_fobj.read())
+                        shutil.copyfileobj(from_fobj, tmp_fobj)
                     hdfs.move(tmp_info.path, to_info.path)
                 except Exception:
                     self.remove(tmp_info)

dvc/tree/webhdfs.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import shutil
 import threading
 from contextlib import contextmanager
 from urllib.parse import urlparse
@@ -134,8 +135,8 @@ def get_file_hash(self, path_info):
 
     def copy(self, from_info, to_info, **_kwargs):
         with self.hdfs_client.read(from_info.path) as reader:
-            content = reader.read()
-        self.hdfs_client.write(to_info.path, data=content)
+            with self.hdfs_client.write(to_info.path) as writer:
+                shutil.copyfileobj(reader, writer)
 
     def move(self, from_info, to_info, mode=None):
         self.hdfs_client.makedirs(to_info.parent.path)