Use msgpack.Unpacker.feed() method

treasure-data · Aug 19, 2024 · df4d6d6 · df4d6d6
1 parent e8d1b95
commit df4d6d6
Showing 1 changed file with 8 additions and 3 deletions.
diff --git a/tdclient/job_api.py b/tdclient/job_api.py
@@ -248,9 +248,14 @@ def job_result_format_each(self, job_id, format, header=False):
             if code != 200:
                 self.raise_error("Get job result failed", res, "")
             if format == "msgpack":
-                unpacker = msgpack.Unpacker(res, raw=False)
-                for row in unpacker:
-                    yield row
+                unpacker = msgpack.Unpacker(raw=False)
+                while True:
+                    buf = res.read(1024 ** 3)
+                    if not buf:
+                        break
+                    unpacker.feed(buf)
+                    for row in unpacker:
+                        yield row
             elif format == "json":
                 for row in codecs.getreader("utf-8")(res):
                     yield json.loads(row)