Releases v0.12.4.1 (#299)

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, 12, 4)
+version_info = (0, 12, 4, 1)
 _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])) + "".join(
     version_info[_num_index + 1 :]

odps/accounts.py

@@ -39,8 +39,9 @@
 
 
 def _get_v4_signature_prefix():
-    if options.signature_prefix is not None:
-        return options.signature_prefix
+    opt_prefix = options.signature_prefix
+    if opt_prefix is not None:
+        return opt_prefix
     return DEFAULT_SIGNATURE_PREFIX
 
 
@@ -558,6 +559,7 @@ def sign_request(self, req, endpoint, region_name=None):
 class CredentialProviderAccount(StsAccount):
     def __init__(self, credential_provider):
         self.provider = credential_provider
+        self._lock = threading.Lock()
         super(CredentialProviderAccount, self).__init__(None, None, None)
 
     def _refresh_credential(self):
@@ -566,15 +568,29 @@ def _refresh_credential(self):
         except:
             credential = self.provider.get_credentials()
 
-        self.access_id = credential.get_access_key_id()
-        self.secret_access_key = credential.get_access_key_secret()
-        self.sts_token = credential.get_security_token()
+        access_id = credential.get_access_key_id()
+        secret_access_key = credential.get_access_key_secret()
+        sts_token = credential.get_security_token()
+
+        with self._lock:
+            if self.access_id and access_id == self.access_id:
+                return
+
+            # invalidate signature date only when token changed
+            self._last_signature_date = None
+
+            self.access_id = access_id
+            self.secret_access_key = secret_access_key
+            self.sts_token = sts_token
 
     def sign_request(self, req, endpoint, region_name=None):
         utils.call_with_retry(self._refresh_credential)
-        return super(CredentialProviderAccount, self).sign_request(
-            req, endpoint, region_name=region_name
-        )
+
+        with self._lock:
+            # need to lock to make sure keys are consistent
+            return super(CredentialProviderAccount, self).sign_request(
+                req, endpoint, region_name=region_name
+            )
 
 
 def from_environments():

odps/models/session/v1.py

@@ -221,7 +221,8 @@ def _check_is_select(sql_statement):
                 RuntimeWarning,
             )
             return False
-        return splited[-1].lower().strip(" \t\r\n(").startswith("select")
+        striped = splited[-1].lower().strip(" \t\r\n(")
+        return striped.startswith("select") or striped.startswith("with")
 
     def _create_internal_instance(self, task=None):
         project_name = self._project.name

odps/models/table.py

@@ -1113,7 +1113,7 @@ def open_writer(
 
         if partition and not isinstance(partition, odps_types.PartitionSpec):
             partition = odps_types.PartitionSpec(partition)
-        if create_partition and not self.exist_partition(create_partition):
+        if create_partition and not self.exist_partition(partition):
             self.create_partition(partition, if_not_exists=True)
 
         tunnel = self._create_table_tunnel(endpoint=endpoint, quota_name=quota_name)
@@ -1269,7 +1269,7 @@ def partitions(self):
 
     @utils.with_wait_argument
     def create_partition(
-        self, partition_spec, if_not_exists=False, async_=False, hints=None
+        self, partition_spec, if_not_exists=False, async_=False, hints=None, **inst_kw
     ):
         """
         Create a partition within the table.
@@ -1282,12 +1282,16 @@ def create_partition(
         :rtype: odps.models.partition.Partition
         """
         return self.partitions.create(
-            partition_spec, if_not_exists=if_not_exists, hints=hints, async_=async_
+            partition_spec,
+            if_not_exists=if_not_exists,
+            hints=hints,
+            async_=async_,
+            **inst_kw
         )
 
     @utils.with_wait_argument
     def delete_partition(
-        self, partition_spec, if_exists=False, async_=False, hints=None
+        self, partition_spec, if_exists=False, async_=False, hints=None, **inst_kw
     ):
         """
         Delete a partition within the table.
@@ -1298,7 +1302,7 @@ def delete_partition(
         :param async_:
         """
         return self.partitions.delete(
-            partition_spec, if_exists=if_exists, hints=hints, async_=async_
+            partition_spec, if_exists=if_exists, hints=hints, async_=async_, **inst_kw
         )
 
     def exist_partition(self, partition_spec):

odps/models/tableio.py

@@ -1550,7 +1550,7 @@ def _format_raw_sql(fmt, args):
                     ),
                 )
                 odps.execute_sql(sql_stmt, **sql_kwargs)
-                tmp_schema = odps.get_table(temp_table_name).table_schema
+                tmp_schema = odps.get_table(temp_table_name, schema=schema).table_schema
                 out_table_schema = cls._resolve_schema(
                     data_schema=tmp_schema,
                     partition=partition,
@@ -1581,7 +1581,7 @@ def _format_raw_sql(fmt, args):
                 partition=partition,
             )
 
-            temp_table = odps.get_table(temp_table_name)
+            temp_table = odps.get_table(temp_table_name, schema=schema)
             union_cols, diff_cols = cls._calc_schema_diff(
                 temp_table.table_schema,
                 target_table.table_schema,

odps/models/tables.py

@@ -133,6 +133,18 @@ def create(
         **kw
     ):
         project_name = self._parent.project.name
+
+        # pass kwargs for instance creation to _run_table_sql
+        inst_kw = {}
+        for key in (
+            "instance_project",
+            "priority",
+            "running_cluster",
+            "unique_identifier_id",
+        ):
+            if key in kw:
+                inst_kw[key] = kw.pop(key)
+
         schema_name = self._get_schema_name()
         sql = Table.gen_create_table_sql(
             table_name,
@@ -152,7 +164,7 @@ def create(
         if storage_tier:
             hints["odps.tiered.storage.enable"] = "true"
         instance = self._run_table_sql(
-            sql, task_name="SQLCreateTableTask", hints=hints, wait=not async_
+            sql, task_name="SQLCreateTableTask", hints=hints, wait=not async_, **inst_kw
         )
         if not async_:
             return self[table_name]

odps/types.py

@@ -327,7 +327,7 @@ def __hash__(self):
     def __eq__(self, other):
         if not isinstance(other, Schema):
             return False
-        return self.names == other.names and self.types == self.types
+        return self.names == other.names and self.types == other.types
 
     def get_type(self, name):
         return self.types[self._name_indexes[utils.to_lower_str(name)]]
@@ -908,6 +908,9 @@ def can_implicit_cast(self, other):
     def can_explicit_cast(self, other):
         return self.can_implicit_cast(other)
 
+    def __reduce__(self):
+        return validate_data_type, (str(self),)
+
     def validate_value(self, val, max_field_size=None):
         # directly return True means without checking
         return True