Releases v0.11.6.3 (#246)

Author: wjsi

odps/_version.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-version_info = (0, 11, 6, 2)
+version_info = (0, 11, 6, 3)
 _num_index = max(idx if isinstance(v, int) else 0
                  for idx, v in enumerate(version_info))
 __version__ = '.'.join(map(str, version_info[:_num_index + 1])) + \

odps/apis/storage_api/tests/test_storage_api.py

@@ -38,8 +38,7 @@ def test_storage_api(storage_api_client):
 
     assert resp.status == Status.OK
     if resp.status != Status.OK:
-        logger.info("Create write session failed")
-        return
+        raise IOError("Create write session failed")
 
     req = SessionRequest(session_id=resp.session_id)
 
@@ -49,7 +48,7 @@ def test_storage_api(storage_api_client):
         assert resp.status == Status.OK
 
         if resp.status != Status.OK:
-            logger.info("Get write session failed")
+            raise IOError("Get write session failed")
             return
 
         if resp.session_status != SessionStatus.NORMAL and resp.session_status != SessionStatus.COMMITTED:
@@ -78,20 +77,21 @@ def test_storage_api(storage_api_client):
         if i == 0:
             suc = writer.write(record_batch.schema.serialize().to_pybytes())
             if not suc:
-                logger.info("write arrow schema failed")
-                break
+                raise IOError("write arrow schema failed")
 
         suc = writer.write(record_batch.serialize().to_pybytes())
         if not suc:
-            logger.info("write arrow record batch failed")
-            break
+            raise IOError("write arrow record batch failed")
 
+    # write EOS given https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
+    suc = writer.write(b'\xff\xff\xff\xff\x00\x00\x00\x00')
+    if not suc:
+        raise IOError("write EOS failed")
     commit_message, suc = writer.finish()
 
     assert suc is True
     if not suc:
-        logger.info("Write rows failed")
-        return
+        raise IOError("Write rows failed")
     else:
         end = time.time()
         logger.info("Write rows cost: " + str(end - start) + "s")
@@ -103,8 +103,7 @@ def test_storage_api(storage_api_client):
     resp = storage_api_client.commit_write_session(req, commit_messages)
 
     if resp.status != Status.OK and resp.status != Status.WAIT:
-        logger.info("Fail to commit write session")
-        return
+        raise IOError("Fail to commit write session")
 
     if resp.status == Status.WAIT:
         req = SessionRequest(session_id=resp.session_id)
@@ -114,8 +113,7 @@ def test_storage_api(storage_api_client):
             assert resp.status == Status.OK
 
             if resp.status != Status.OK:
-                logger.info("Get write session failed")
-                return
+                raise IOError("Get write session failed")
 
             if resp.session_status != SessionStatus.NORMAL and resp.session_status != SessionStatus.COMMITTED:
                 logger.info("Wait...")
@@ -131,17 +129,15 @@ def test_storage_api(storage_api_client):
     resp = storage_api_client.create_read_session(req)
 
     if resp.status != Status.OK and resp.status != Status.WAIT:
-        logger.info("create read session failed")
-        return
+        raise IOError("create read session failed")
 
     req = SessionRequest(session_id=resp.session_id)
 
     while True:
         resp = storage_api_client.get_read_session(req)
 
         if resp.status != Status.OK:
-            logger.info("get read session failed")
-            return
+            raise IOError("get read session failed")
 
         if resp.session_status == SessionStatus.INIT:
             logger.info("Wait...")
@@ -168,8 +164,7 @@ def test_storage_api(storage_api_client):
 
         reader.close()
         if reader.get_status() != Status.OK:
-            logger.info("Read rows failed")
-            return
+            raise IOError("Read rows failed")
 
         end = time.time()
         logger.info("Read rows cost (index " + str(i) + "): " + str(end - start) + "s")

odps/core.py

@@ -14,6 +14,7 @@
 
 import copy
 import functools
+import hashlib
 import json  # noqa: F401
 import os
 import random
@@ -2357,6 +2358,18 @@ def _create_mcqa_session(
             task=task, session_project=project, session_name=session_name
         )
 
+    def _get_mcqa_session_file(self):
+        try:
+            dir_name = utils.build_pyodps_dir()
+            if not os.path.exists(dir_name):
+                os.makedirs(dir_name)
+            access_id_digest = hashlib.md5(utils.to_binary(self.account.access_id)).hexdigest()
+            return os.path.join(
+                dir_name, "mcqa-session-" + access_id_digest
+            )
+        except:
+            return None
+
     def run_sql_interactive(self, sql, hints=None, **kwargs):
         """
         Run SQL query in interactive mode (a.k.a MaxCompute QueryAcceleration).
@@ -2370,10 +2383,29 @@ def run_sql_interactive(self, sql, hints=None, **kwargs):
         task_name = kwargs.pop('task_name', None)
         service_startup_timeout = kwargs.pop('service_startup_timeout', 60)
         force_reattach = kwargs.pop('force_reattach', False)
-        if self._default_session != None:
+
+        session_file_name = self._get_mcqa_session_file()
+        if (
+            self._default_session is None
+            and session_file_name
+            and os.path.exists(session_file_name)
+        ):
+            try:
+                with open(session_file_name, "r") as session_file:
+                    session_info = json.loads(session_file.read())
+                instance_obj = self.get_instance(session_info.pop("id"))
+                session_project = self.get_project(session_info.pop("session_project_name"))
+                self._default_session_name = session_info["session_name"]
+                self._default_session = models.SessionInstance.from_instance(
+                    instance_obj, session_project=session_project, **session_info
+                )
+            except:
+                pass
+
+        if self._default_session is not None:
             try:
                 cached_is_running = self._default_session.is_running()
-            except BaseException:
+            except:
                 pass
         if (
             force_reattach
@@ -2385,6 +2417,15 @@ def run_sql_interactive(self, sql, hints=None, **kwargs):
             self._default_session = self._attach_mcqa_session(service_name, task_name=task_name)
             self._default_session.wait_for_startup(0.1, service_startup_timeout, max_interval=1)
             self._default_session_name = service_name
+
+            if session_file_name:
+                try:
+                    with open(session_file_name, "w") as session_file:
+                        session_file.write(
+                            json.dumps(self._default_session._extract_json_info())
+                        )
+                except:
+                    pass
         return self._default_session.run_sql(sql, hints, **kwargs)
 
     @utils.deprecated(

odps/models/__init__.py

@@ -25,6 +25,7 @@
 from .function import Function
 from .resources import Resources
 from .resource import *
+from .session import InSessionInstance, SessionInstance
 from .tenant import Tenant
 from .volumes import *
 from .volume_parted import PartedVolume, VolumePartition

odps/models/session.py

@@ -147,6 +147,20 @@ def __init__(self, **kw):
         self._session_name = kw.pop("session_name", "")
         super(SessionInstance, self).__init__(**kw)
 
+    @classmethod
+    def from_instance(cls, instance, **kw):
+        return SessionInstance(
+            name=instance.id, parent=instance.parent, client=instance._client, **kw
+        )
+
+    def _extract_json_info(self):
+        return {
+            "id": self.id,
+            "session_project_name": self._project.name,
+            "session_task_name": self._task_name,
+            "session_name": self._session_name,
+        }
+
     def wait_for_startup(self, interval=1, timeout=-1, retry=True, max_interval=None):
         """
         Wait for the session to startup(status changed to RUNNING).

odps/models/tests/test_session.py

@@ -358,3 +358,11 @@ def test_fallback_policy():
     assert default_policy.get_mode_from_exception(
         errors.SQAQueryTimedout("QueryTimedout")
     ) is FallbackMode.OFFLINE
+
+
+def test_reuse_session(odps):
+    inst = odps.run_sql_interactive("select * from dual", force_reattach=True)
+    inst.wait_for_completion()
+    new_odps = ODPS(account=odps.account, project=odps.project, endpoint=odps.endpoint)
+    new_inst = new_odps.run_sql_interactive("select * from dual")
+    assert inst.id == new_inst.id

odps/tunnel/io/reader.py

@@ -508,7 +508,7 @@ def to_pandas(self):
 
         if not batches:
             return self._arrow_schema.empty_table().to_pandas()
-        return pd.concat(batches, axis=0)
+        return pd.concat(batches, axis=0, ignore_index=True)
 
     def __enter__(self):
         return self

odps/tunnel/io/writer.py

@@ -438,15 +438,7 @@ def _reset_writer(self, write_response):
         self._crc.reset()
 
     def _send_buffer(self):
-        def gen():  # synchronize chunk upload
-            data = self._buffer.getvalue()
-            chunk_size = options.chunk_size
-            while data:
-                to_send = data[:chunk_size]
-                data = data[chunk_size:]
-                yield to_send
-
-        return self._request_callback(self._block_id, gen())
+        return self._request_callback(self._block_id, self._buffer.getvalue())
 
     def _flush(self):
         self._write_finish_tags()
@@ -543,20 +535,20 @@ def _re_init(self, output):
         self._output = output
 
     def _write_chunk_size(self):
-        self._write_unint32(self._chunk_size)
+        self._write_uint32(self._chunk_size)
 
-    def _write_unint32(self, val):
+    def _write_uint32(self, val):
         data = struct.pack("!I", utils.long_to_uint(val))
         self._output.write(data)
 
     def _write_chunk(self, buf):
         self._output.write(buf)
+        self._crc.update(buf)
         self._crccrc.update(buf)
         self._cur_chunk_size += len(buf)
         if self._cur_chunk_size >= self._chunk_size:
-            self._crc.update(buf)
             checksum = self._crc.getvalue()
-            self._write_unint32(checksum)
+            self._write_uint32(checksum)
             self._crc.reset()
             self._cur_chunk_size = 0
 
@@ -602,7 +594,7 @@ def write(self, data):
 
                 if tp == pa.timestamp("ms") or tp == pa.timestamp("ns"):
                     if self._schema[name].type == types.timestamp_ntz:
-                        column_dict[name] = self._localize_timezone(column_dict, "UTC")
+                        column_dict[name] = self._localize_timezone(column_dict[name], "UTC")
                     else:
                         column_dict[name] = self._localize_timezone(column_dict[name])
 
@@ -633,7 +625,7 @@ def write(self, data):
 
     def _write_finish_tags(self):
         checksum = self._crccrc.getvalue()
-        self._write_unint32(checksum)
+        self._write_uint32(checksum)
         self._crccrc.reset()
 
     def flush(self):
@@ -723,15 +715,7 @@ def _reset_writer(self):
         self._crc.reset()
 
     def _send_buffer(self):
-        def gen():  # synchronize chunk upload
-            data = self._buffer.getvalue()
-            chunk_size = options.chunk_size
-            while data:
-                to_send = data[:chunk_size]
-                data = data[chunk_size:]
-                yield to_send
-
-        return self._request_callback(self._block_id, gen())
+        return self._request_callback(self._block_id, self._buffer.getvalue())
 
     def _flush(self):
         self._write_finish_tags()

odps/tunnel/tabletunnel.py

@@ -355,6 +355,14 @@ def _init(self):
     def reload(self):
         self._create_or_reload_session(reload=True)
 
+    @classmethod
+    def _iter_data_in_batches(cls, data):
+        pos = 0
+        chunk_size = options.chunk_size
+        while pos < len(data):
+            yield data[pos: pos + chunk_size]
+            pos += chunk_size
+
     def _open_writer(
         self,
         block_id=None,
@@ -387,9 +395,15 @@ def _open_writer(
             @_wrap_upload_call(self.id)
             def upload_block(blockid, data):
                 params['blockid'] = blockid
-                return utils.call_with_retry(
-                    self._client.put, url, data=data, params=params, headers=headers
-                )
+
+                def upload_func():
+                    if isinstance(data, (bytes, bytearray)):
+                        to_upload = self._iter_data_in_batches(data)
+                    else:
+                        to_upload = data
+                    return self._client.put(url, data=to_upload, params=params, headers=headers)
+
+                return utils.call_with_retry(upload_func)
 
             if writer_cls is ArrowWriter:
                 writer_cls = BufferedArrowWriter