upgrade to 0.8.2

Author: 继盛

odps/_version.py

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

@@ -136,6 +136,9 @@ def _do_POST(self):
                 self.end_headers()
                 return
 
+            self._sign(postvars)
+
+        def _sign(self, postvars):
             if self.server._token is not None:
                 auth = self.headers.get('Authorization')
                 if not auth:
@@ -267,3 +270,16 @@ def sign_request(self, req, endpoint):
             LOG.debug('headers after signing: ' + repr(req.headers))
         else:
             raise SignServerError('Sign server returned error code: %d' % resp.status_code, resp.status_code)
+
+
+class BearerTokenAccount(BaseAccount):
+    def __init__(self, token):
+        self._token = token
+
+    @property
+    def token(self):
+        return self._token
+
+    def sign_request(self, req, endpoint):
+        req.headers['x-odps-bearer-token'] = self._token
+        LOG.debug('headers after signing: ' + repr(req.headers))

odps/config.py

@@ -25,7 +25,7 @@
 
 DEFAULT_CHUNK_SIZE = 1496
 DEFAULT_CONNECT_RETRY_TIMES = 4
-DEFAULT_CONNECT_TIMEOUT = 5
+DEFAULT_CONNECT_TIMEOUT = 10
 DEFAULT_READ_TIMEOUT = 120
 DEFAULT_POOL_CONNECTIONS = 10
 DEFAULT_POOL_MAXSIZE = 10
@@ -329,7 +329,7 @@ def validate(x):
 options.register_option('allow_antique_date', False)
 options.register_option('user_agent_pattern', '$pyodps_version $python_version $os_version')
 options.register_option('log_view_host', None)
-options.register_option('log_view_hours', 24, validator=is_integer)
+options.register_option('log_view_hours', 24 * 30, validator=is_integer)
 options.register_option('api_proxy', None)
 options.register_option('data_proxy', None)
 options.redirect_option('tunnel_proxy', 'data_proxy')

odps/df/backends/odpssql/compiler.py

@@ -1398,12 +1398,11 @@ def visit_join(self, expr):
         self._ctx.add_expr_compiled(expr, from_clause)
 
     def visit_union(self, expr):
-        if expr._distinct:
-            raise CompileError("Distinct union is not supported here.")
-
+        union_type = 'UNION ALL' if not expr._distinct else 'UNION'
         left_compiled, right_compiled = tuple(self._sub_compiles[expr])
 
-        from_clause = '{0} \nUNION ALL\n{1}'.format(left_compiled, utils.indent(right_compiled, self._indent_size))
+        from_clause = '{0} \n{1}\n{2}'.format(left_compiled, union_type,
+                                              utils.indent(right_compiled, self._indent_size))
 
         compiled = from_clause
         if not self._union_no_alias.get(expr, False):

odps/df/backends/odpssql/tests/test_compiler.py

@@ -2546,6 +2546,20 @@ def testUnion(self):
 
         self.assertEqual(to_str(expected), to_str(ODPSEngine(self.odps).compile(expr, prettify=False)))
 
+        expr = e.union(e3['const', 'id', 'name'], distinct=True)
+
+        expected = "SELECT * \n" \
+                   "FROM (\n" \
+                   "  SELECT t1.`name`, t1.`id`, 'cst' AS `const` \n" \
+                   "  FROM mocked_project.`pyodps_test_expr_table` t1 \n" \
+                   "  UNION\n" \
+                   "    SELECT t2.`name`, SUM(t2.`id`) AS `id`, 'cst' AS `const` \n" \
+                   "    FROM mocked_project.`pyodps_test_expr_table1` t2 \n" \
+                   "    GROUP BY t2.`name`\n" \
+                   ") t3"
+
+        self.assertEqual(to_str(expected), to_str(ODPSEngine(self.odps).compile(expr, prettify=False)))
+
     def testAliases(self):
         df = self.expr
         df = df[(df.id == 1) | (df.id == 2)].exclude(['fid'])

odps/df/backends/odpssql/types.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 # Copyright 1999-2017 Alibaba Group Holding Ltd.
-# 
+#
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
-# 
+#
 #      http://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -18,6 +18,7 @@
 from .... import types as odps_types
 from ... import types as df_types
 from ....compat import six
+from ....config import options
 
 
 _odps_to_df_types = {
@@ -46,6 +47,19 @@
     df_types.binary: odps_types.string,
 }
 
+_df_to_odps_types2 = {
+    df_types.int8: odps_types.tinyint,
+    df_types.int16: odps_types.smallint,
+    df_types.int32: odps_types.int_,
+    df_types.int64: odps_types.bigint,
+    df_types.float32: odps_types.float_,
+    df_types.float64: odps_types.double,
+    df_types.boolean: odps_types.boolean,
+    df_types.string: odps_types.string,
+    df_types.datetime: odps_types.datetime,
+    df_types.binary: odps_types.binary,
+}
+
 
 def odps_type_to_df_type(odps_type):
     if isinstance(odps_type, six.string_types):
@@ -74,11 +88,15 @@ def odps_schema_to_df_schema(odps_schema):
 
 
 def df_type_to_odps_type(df_type):
+    if options.sql.use_odps2_extension:
+        df_to_odps_types = _df_to_odps_types2
+    else:
+        df_to_odps_types = _df_to_odps_types
     if isinstance(df_type, six.string_types):
         df_type = df_types.validate_data_type(df_type)
 
-    if df_type in _df_to_odps_types:
-        return _df_to_odps_types[df_type]
+    if df_type in df_to_odps_types:
+        return df_to_odps_types[df_type]
     elif df_type == df_types.decimal:
         return odps_types.Decimal()
     elif isinstance(df_type, df_types.List):

odps/df/backends/tests/test_mixed_engine.py

@@ -18,12 +18,12 @@
 import time
 import uuid
 
-from odps.tests.core import tn, pandas_case
+from odps import types as odps_types
+from odps.tests.core import tn, pandas_case, odps2_typed_case
 from odps.df.backends.tests.core import TestBase
 from odps.config import options
 from odps.compat import unittest
 from odps.models import Schema, Instance
-from odps.errors import ODPSError
 from odps.df.backends.engine import MixedEngine
 from odps.df.backends.odpssql.engine import ODPSSQLEngine
 from odps.df.backends.pd.engine import PandasEngine
@@ -194,6 +194,31 @@ def testPandasPersist(self):
         finally:
             self.odps.delete_table(tmp_table_name)
 
+    @odps2_typed_case
+    def testPandasPersistODPS2(self):
+        import pandas as pd
+        import numpy as np
+
+        data_int8 = np.random.randint(0, 10, (1,), dtype=np.int8)
+        data_int16 = np.random.randint(0, 10, (1,), dtype=np.int16)
+        data_int32 = np.random.randint(0, 10, (1,), dtype=np.int32)
+        data_int64 = np.random.randint(0, 10, (1,), dtype=np.int64)
+        data_float32 = np.random.random((1,)).astype(np.float32)
+        data_float64 = np.random.random((1,)).astype(np.float64)
+
+        df = DataFrame(pd.DataFrame(dict(data_int8=data_int8, data_int16=data_int16,
+                                         data_int32=data_int32, data_int64=data_int64,
+                                         data_float32=data_float32, data_float64=data_float64)))
+        tmp_table_name = tn('pyodps_test_mixed_persist_odps2_types')
+
+        self.odps.delete_table(tmp_table_name, if_exists=True)
+        df.persist(tmp_table_name, lifecycle=1, drop_table=True, odps=self.odps)
+
+        t = self.odps.get_table(tmp_table_name)
+        expected_types = [odps_types.tinyint, odps_types.smallint, odps_types.int_,
+                          odps_types.bigint, odps_types.float_, odps_types.double]
+        self.assertEqual(expected_types, t.schema.types)
+
     def testExecuteCacheTable(self):
         df = self.odps_df[self.odps_df.name == 'name1']
         result = df.execute().values

odps/models/instance.py

@@ -841,6 +841,8 @@ def open_reader(self, *args, **kwargs):
                        if false, use conventional routine.
                        if absent, `options.tunnel.use_instance_tunnel` will be used and automatic fallback
                        is enabled.
+        :param limit: if True, enable the limitation
+        :type limit: bool
         :param reopen: the reader will reuse last one, reopen is true means open a new reader.
         :type reopen: bool
         :param endpoint: the tunnel service URL

odps/models/instances.py

@@ -133,7 +133,7 @@ def _get_submit_instance_content(cls, job):
         return Instance.AnonymousSubmitInstance(job=job).serialize()
 
     def create(self, xml=None, job=None, task=None, priority=None, running_cluster=None,
-               headers=None, create_callback=None):
+               headers=None, create_callback=None, encoding=None):
         if xml is None:
             job = self._create_job(job=job, task=task, priority=priority,
                                    running_cluster=running_cluster)
@@ -155,6 +155,8 @@ def create(self, xml=None, job=None, task=None, priority=None, running_cluster=N
         if create_callback is not None:
             create_callback(instance_id)
 
+        if encoding is not None:
+            resp.encoding = encoding
         body = resp.text
         if body:
             instance_result = Instance.InstanceResult.parse(self._client, resp)

odps/models/table.py

@@ -270,10 +270,11 @@ def gen_create_table_sql(table_name, table_schema, comment=None, if_not_exists=F
         project = utils.to_text(project)
         comment = utils.to_text(comment)
 
-        store_as = kw.get('store_as')
+        stored_as = kw.get('stored_as')
+        external_stored_as = kw.get('external_stored_as')
         storage_handler = kw.get('storage_handler')
 
-        buf.write(u'CREATE%s TABLE ' % (' EXTERNAL' if storage_handler or store_as else ''))
+        buf.write(u'CREATE%s TABLE ' % (' EXTERNAL' if storage_handler or external_stored_as else ''))
         if if_not_exists:
             buf.write(u'IF NOT EXISTS ')
         if project is not None:
@@ -319,11 +320,11 @@ def write_columns(col_array):
         serde_properties = kw.get('serde_properties')
         location = kw.get('location')
         resources = kw.get('resources')
-        if storage_handler or store_as:
+        if storage_handler or external_stored_as:
             if storage_handler:
                 buf.write("STORED BY '%s'\n" % escape_odps_string(storage_handler))
             else:
-                buf.write("STORED AS %s\n" % escape_odps_string(store_as))
+                buf.write("STORED AS %s\n" % escape_odps_string(external_stored_as))
             if serde_properties:
                 buf.write('WITH SERDEPROPERTIES (\n')
                 for idx, k in enumerate(serde_properties):
@@ -344,6 +345,9 @@ def write_columns(col_array):
             if hub_lifecycle is not None:
                 buf.write(u' HUBLIFECYCLE %s\n' % hub_lifecycle)
 
+        if stored_as:
+            buf.write("STORED AS %s\n" % escape_odps_string(stored_as))
+
         return buf.getvalue().strip()
 
     def get_ddl(self, with_comments=True, if_not_exists=False):