Upgrade to 0.10.2 (#135)

Author: hekaisheng

cupid/config.py

@@ -33,4 +33,4 @@
 
 # mars app config
 options.register_option('cupid.image_prefix', None)
-options.register_option('cupid.image_version', '0.9.4rc3')
+options.register_option('cupid.image_version', 'v2.0')

docs/source/base-sqlalchemy.rst

@@ -17,6 +17,13 @@ PyODPS 支持集成 SQLAlchemy，可以使用 SQLAlchemy 查询 MaxCompute 数
    from sqlalchemy import create_engine
    engine = create_engine('odps://<access_id>:<access_key>@<project>')
 
+要在连接串中指定 ``endpoint``，可以按如下方式：
+
+.. code-block:: python
+
+   from sqlalchemy import create_engine
+   engine = create_engine('odps://<access_id>:<access_key>@<project>/?endpoint=<endpoint>')
+
 这里把 ``<access_id>`` 等替换成相应的账号。
 
 对于已有的 ODPS 对象 ``o`` ，调用 ``o.to_global()`` 设为全局账号后，在连接串中就不需要指定了。

odps/_version.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-version_info = (0, 10, 1)
+version_info = (0, 10, 2)
 _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/accounts.py

@@ -293,26 +293,36 @@ def sign_request(self, req, endpoint):
 
 
 class BearerTokenAccount(BaseAccount):
-    def __init__(self, token, expired_hours=5):
+    def __init__(self, token, expired_hours=5, get_bearer_token_fun=None):
         self._token = token
         self._last_modified_time = None
         self._expired_time = timedelta(hours=expired_hours)
+        self._get_bearer_token = get_bearer_token_fun or self.get_bearer_token
 
-    def _check_bearer_token(self):
+    @staticmethod
+    def get_bearer_token():
         from cupid import context
 
         cupid_context = context()
         if cupid_context is None:
             return
 
+        return cupid_context.get_bearer_token()
+
+    def _check_bearer_token(self):
         t = datetime.now()
         if self._last_modified_time is None:
-            token = cupid_context.get_bearer_token()
+            token = self._get_bearer_token()
+            if token is None:
+                return
             if token != self._token:
                 self._token = token
                 self._last_modified_time = datetime.now()
         elif (t - self._last_modified_time) > self._expired_time:
-            self._token = cupid_context.get_bearer_token()
+            token = self._get_bearer_token()
+            if token is None:
+                return
+            self._token = token
             self._last_modified_time = datetime.now()
 
     @property
@@ -321,5 +331,8 @@ def token(self):
 
     def sign_request(self, req, endpoint):
         self._check_bearer_token()
+        url = req.url[len(endpoint):]
+        url_components = urlparse(unquote(url), allow_fragments=False)
+        self._build_canonical_str(url_components, req)
         req.headers['x-odps-bearer-token'] = self._token
         LOG.debug('headers after signing: ' + repr(req.headers))

odps/core.py

@@ -130,6 +130,10 @@ def __getstate__(self):
         return params
 
     def __setstate__(self, state):
+        if 'secret_access_key' in state:
+            # if `secret_access_key` in state
+            self._init(**state)
+            return
         try:
             bearer_token = os.environ['ODPS_BEARER_TOKEN']
             state['project'] = os.environ['ODPS_PROJECT_NAME']
@@ -141,6 +145,9 @@ def __setstate__(self, state):
         except KeyError:
             self._init(**state)
 
+    def __mars_tokenize__(self):
+        return self.__getstate__()
+
     @classmethod
     def _from_account(cls, account, project, endpoint=DEFAULT_ENDPOINT,
                       tunnel_endpoint=None, **kwargs):
@@ -1782,6 +1789,14 @@ def attach_session(self, session_name, taskname=None, hints=None):
                                         session_project=project,
                                         session_name=session_name)
 
+    def default_session(self):
+        """
+        Attach to the default session of your project.
+
+        :return: A SessionInstance you may execute select tasks within.
+        """
+        return self.attach_session('public.default')
+
     def create_session(self, session_worker_count, session_worker_memory,
             session_name=None, worker_spare_span=None, taskname=None, hints=None):
         """

odps/dbapi.py

@@ -139,13 +139,18 @@ def description(self):
             return
         if self._description is None:
             self._check_download_session()
+            self._description = []
             if self._download_session is not None:
-                self._description = []
                 for col in self._download_session.schema.columns:
                     self._description.append((
                         col.name, col.type.name,
                         None, None, None, None, True
                     ))
+            else:
+                self._description.append((
+                    '_c0', 'string', None, None,
+                    None, None, True
+                ))
         return self._description
 
     @staticmethod
@@ -208,8 +213,18 @@ def _check_download_session(self):
                 # not select, cannot create session
                 self._download_session = None
 
+    def _fetch_non_select(self):
+        # not select
+        # just return reader.raw
+        with self._instance.open_reader() as reader:
+            return [(reader.raw,)]
+
     def _fetch(self, size):
         self._check_download_session()
+
+        if self._download_session is None:
+            return self._fetch_non_select()
+
         results = []
         i = 0
         while size == -1 or i < size:

odps/mars_extension/__init__.py

@@ -49,5 +49,15 @@
     pass
 
 
+# register filesystem
+try:
+    from mars.filesystem import file_systems
+    from .filesystem import VolumeFileSystem
+
+    file_systems['odps'] = VolumeFileSystem
+except ImportError:
+    pass
+
+
 INTERNAL_PATTERN = '\/[^\.]+\.[^\.-]+\.[^\.-]+\-[^\.-]+\.'
 

odps/mars_extension/core.py

@@ -145,7 +145,7 @@ def create_mars_cluster(odps, worker_num=1, worker_cpu=8, worker_mem=32, cache_m
     disk_size = int(disk_size * 1024 ** 3)
     scheduler_mem = int(scheduler_mem * 1024 ** 3) if scheduler_mem else None
     web_mem = int(web_mem * 1024 ** 3) if web_mem else None
-    notebook_mem = int(notebook_mem * 1024 ** 3) if web_mem else None
+    notebook_mem = int(notebook_mem * 1024 ** 3) if notebook_mem else None
 
     kw = dict(worker_num=worker_num, worker_cpu=worker_cpu, worker_mem=worker_mem,
               worker_cache_mem=cache_mem, min_worker_num=min_worker_num,
@@ -176,7 +176,7 @@ def to_mars_dataframe(odps, table_name, shape=None, partition=None, chunk_bytes=
     :param use_arrow_dtype: read to arrow dtype. Reduce memory in some saces.
     :return: Mars DataFrame.
     """
-    from cupid import context
+    from cupid.runtime import RuntimeContext
     from .dataframe import read_odps_table
     from ..utils import init_progress_ui
 
@@ -204,7 +204,7 @@ def to_mars_dataframe(odps, table_name, shape=None, partition=None, chunk_bytes=
 
     # get dataframe's shape
     if shape is None:
-        if calc_nrows and context() is None:
+        if calc_nrows and not RuntimeContext.is_context_ready():
             # obtain count
             if partition is None:
                 odps_df = data_src.to_df()

odps/mars_extension/dataframe/datasource.py

@@ -125,12 +125,14 @@ def tile(cls, op):
         from odps import ODPS
         from odps.accounts import BearerTokenAccount
         from cupid import CupidSession, context
+        from cupid.runtime import RuntimeContext
         from mars.context import get_context
 
-        cupid_ctx = context()
-        if cupid_ctx is None:
+        if not RuntimeContext.is_context_ready():
             raise SystemError('No Mars cluster found, please create via `o.create_mars_cluster`.')
 
+        cupid_ctx = context()
+
         bearer_token = cupid_ctx.get_bearer_token()
         account = BearerTokenAccount(bearer_token)
         project = os.environ.get('ODPS_PROJECT_NAME', None)
@@ -190,8 +192,6 @@ def tile(cls, op):
         logger.warning('Estimated chunk rows: %r', est_chunk_rows)
 
         out_chunks = []
-        # Ignore add_offset at this time.
-        op._add_offset = False
 
         if len(download_session.splits) == 0:
             logger.debug('Table {} has no data'.format(op.table_name))

odps/mars_extension/dataframe/datastore.py

@@ -105,11 +105,12 @@ def tile(cls, op):
         from odps import ODPS
         from odps.accounts import BearerTokenAccount
         from cupid import CupidSession, context
+        from cupid.runtime import RuntimeContext
         from mars.dataframe.utils import build_concatenated_rows_frame
 
-        cupid_ctx = context()
-        if cupid_ctx is None:
+        if not RuntimeContext.is_context_ready():
             raise SystemError('No Mars cluster found, please create via `o.create_mars_cluster`.')
+        cupid_ctx = context()
 
         bearer_token = cupid_ctx.get_bearer_token()
         account = BearerTokenAccount(bearer_token)
@@ -146,7 +147,7 @@ def tile(cls, op):
         # build commit tree
         combine_size = 8
         chunks = out_chunks
-        while len(chunks) > combine_size:
+        while len(chunks) >= combine_size:
             new_chunks = []
             for i in range(0, len(chunks), combine_size):
                 chks = chunks[i: i + combine_size]