/** * Copyright (c) 2021 OceanBase * OceanBase CE is licensed under Mulan PubL v2. * You can use this software according to the terms and conditions of the Mulan PubL v2. * You may obtain a copy of Mulan PubL v2 at: * http://license.coscl.org.cn/MulanPubL-2.0 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. * See the Mulan PubL v2 for more details. */ #define USING_LOG_PREFIX SQL_SESSION #include "sql/session/ob_session_val_map.h" #include "lib/hash/ob_hashutils.h" #include "share/schema/ob_schema_struct.h" using namespace oceanbase; using namespace sql; using namespace common; using namespace share; using namespace share::schema; ObSessionValMap::ObSessionValMap() : block_allocator_(SMALL_BLOCK_SIZE, common::OB_MALLOC_NORMAL_BLOCK_SIZE, ObMalloc(ObModIds::OB_SQL_SESSION_VAR_MAP)), var_name_val_map_allocer_(SMALL_BLOCK_SIZE, ObWrapperAllocator(&block_allocator_)), str_buf1_(ObModIds::OB_SQL_SESSION_VAR_MAP, 64 * 1024), str_buf2_(ObModIds::OB_SQL_SESSION_VAR_MAP, 64 * 1024), current_buf_index_(0), bucket_allocator_(ObModIds::OB_SQL_SESSION_VAR_MAP), bucket_allocator_wrapper_(&bucket_allocator_), str_buf_free_threshold_(0), next_free_mem_point_(str_buf_free_threshold_) { str_buf_[0] = &str_buf1_; str_buf_[1] = &str_buf2_; } ObSessionValMap::ObSessionValMap(const int64_t block_size, const ObWrapperAllocator &block_allocator, const int64_t tenant_id) : block_allocator_(SMALL_BLOCK_SIZE, common::OB_MALLOC_NORMAL_BLOCK_SIZE, ObMalloc(ObMemAttr(tenant_id, ObModIds::OB_SQL_SESSION_VAR_MAP))), var_name_val_map_allocer_(block_size, block_allocator), str_buf1_(tenant_id, ObModIds::OB_SQL_SESSION_VAR_MAP), str_buf2_(tenant_id, ObModIds::OB_SQL_SESSION_VAR_MAP), current_buf_index_(0), bucket_allocator_(ObMemAttr(tenant_id, ObModIds::OB_SQL_SESSION_VAR_MAP)), bucket_allocator_wrapper_(&bucket_allocator_), str_buf_free_threshold_(0), next_free_mem_point_(str_buf_free_threshold_) { str_buf_[0] = &str_buf1_; str_buf_[1] = &str_buf2_; } ObSessionValMap::~ObSessionValMap() { } int ObSessionValMap::init(int64_t free_threshold, int64_t bucket_num, common::ObWrapperAllocator *bucket_allocator) { int ret = OB_SUCCESS; str_buf_free_threshold_ = free_threshold; next_free_mem_point_ = str_buf_free_threshold_; if (NULL != bucket_allocator) { bucket_allocator_wrapper_.set_alloc(bucket_allocator); } if (OB_FAIL(map_.create(bucket_num, &var_name_val_map_allocer_, &bucket_allocator_wrapper_))) { _OB_LOG(WARN, "fail to create hashmap:ret[%d]", ret); } return ret; } void ObSessionValMap::reuse() { map_.clear(); str_buf1_.reset(); str_buf2_.reset(); } void ObSessionValMap::reset() { reuse(); map_.destroy(); var_name_val_map_allocer_.reset(); bucket_allocator_.reset(); block_allocator_.reset(); } // 成功则返回OB_SUCCESS // 其他返回码均表示不成功 int ObSessionValMap::set_refactored(const common::ObString &name, const ObSessionVariable &sess_var) { int ret = 0; ObString tmp_name; ObSessionVariable tmp_sess_var; if (current_buf_index_ >= 2 || current_buf_index_ < 0) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("invalid current_buf_index_", K(current_buf_index_), K(name), K(sess_var)); } else if (OB_ISNULL(str_buf_[current_buf_index_])) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("invalid internal args", KP(str_buf_[current_buf_index_]), K(name), K(sess_var)); } else if (OB_FAIL(free_mem())) { LOG_WARN("fail to free mem", K(ret)); } else if (OB_FAIL(str_buf_[current_buf_index_]->write_string(name, &tmp_name))) { LOG_WARN("fail to write name to string_buf", K(ret), K(current_buf_index_), K(name), K(sess_var)); } else if (OB_FAIL(str_buf_[current_buf_index_]->write_obj(sess_var.value_, &tmp_sess_var.value_))) { LOG_WARN("fail to write obj to string_buf", K(ret), K(current_buf_index_), K(name), K(sess_var)); } else if (FALSE_IT(tmp_sess_var.meta_ = sess_var.meta_)) { } else if (OB_FAIL(map_.set_refactored(tmp_name, tmp_sess_var, 1))) { LOG_WARN("fail to add item to hash", K(ret), K(tmp_name), K(tmp_sess_var)); } return ret; } int ObSessionValMap::get_refactored(const ObString &name, ObSessionVariable &sess_var) const { return map_.get_refactored(name, sess_var); } const ObSessionVariable *ObSessionValMap::get(const common::ObString &name) const { return map_.get(name); } int ObSessionValMap::erase_refactored(const ObString &key, ObSessionVariable *sess_var) { return map_.erase_refactored(key, sess_var); } int ObSessionValMap::assign(const ObSessionValMap &other) { int ret = OB_SUCCESS; reuse(); if (other.size() > 0) { for (VarNameValMap::const_iterator iter = other.map_.begin(); (OB_SUCCESS ==ret ) && iter != other.map_.end(); iter++) { const ObString &key = iter->first; const ObSessionVariable &value = iter->second; ret = set_refactored(key, value); } } return ret; } int ObSessionValMap::free_mem() { int ret = OB_SUCCESS; if (current_buf_index_ >= 2 || current_buf_index_ < 0) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("invalid current_buf_index_", K(current_buf_index_)); } else if (OB_ISNULL(str_buf_[current_buf_index_])) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("invalid internal args", KP(str_buf_[current_buf_index_])); } else if (str_buf_[current_buf_index_]->used() > next_free_mem_point_) { _OB_LOG(DEBUG, "string buf[%d] is large[%ld], switch string buf", current_buf_index_, str_buf_[current_buf_index_]->used()); int32_t next_index = (current_buf_index_ == 0 ? 1 : 0); int64_t count = 0; if (map_.size() > 0) { VarNameValMap::iterator iter; for (iter = map_.begin(); OB_SUCC(ret) && iter != map_.end(); ++ iter) { count ++; if (OB_FAIL(str_buf_[next_index]->write_string(iter->first, &(iter->first)))) { _OB_LOG(WARN, "fail to write string:ret[%d]", ret); } else if (OB_FAIL(str_buf_[next_index]->write_obj(iter->second.value_, &(iter->second.value_)))) { _OB_LOG(WARN, "fail to write obj:ret[%d]", ret); } } } _OB_LOG(DEBUG, "[%ld] val migrate", count); if (OB_SUCC(ret)) { str_buf_[current_buf_index_]->reuse(); current_buf_index_ = next_index; next_free_mem_point_ = std::max(2 * str_buf_[next_index]->used(), str_buf_free_threshold_); _OB_LOG(DEBUG, "next_free_mem_point_[%ld]", next_free_mem_point_); } } return ret; } DEFINE_SERIALIZE(ObSessionValMap) { int ret = OB_SUCCESS; int64_t num_of_entries = size(); OB_UNIS_ENCODE(num_of_entries); if (OB_SUCC(ret) && num_of_entries > 0) { for (VarNameValMap::const_iterator iter = map_.begin(); (OB_SUCC(ret)) && iter != map_.end(); iter++) { const ObString &key = iter->first; const ObSessionVariable &value = iter->second; OB_UNIS_ENCODE(key); OB_UNIS_ENCODE(value.meta_); OB_UNIS_ENCODE(value.value_); } } return ret; } DEFINE_DESERIALIZE(ObSessionValMap) { int ret = OB_SUCCESS; int64_t num_of_entries = 0; reuse(); OB_UNIS_DECODE(num_of_entries); ObString key; ObSessionVariable value; if (num_of_entries > 0 && !map_.created()) { if (OB_FAIL(init(256 * 1024, 64, NULL))) { SQL_ENG_LOG(WARN, "init map failed", K(ret)); } } for (int64_t i = 0; (OB_SUCC(ret)) && (i < num_of_entries); ++i) { key.reset(); value.reset(); OB_UNIS_DECODE(key); OB_UNIS_DECODE(value.meta_); OB_UNIS_DECODE(value.value_); ret = this->set_refactored(key,value); if (OB_FAIL(ret)) { SQL_ENG_LOG(WARN, "Insert value into map failed", K(ret)); } } return ret; } DEFINE_GET_SERIALIZE_SIZE(ObSessionValMap) { int64_t len = 0; int64_t num_of_entries = size(); OB_UNIS_ADD_LEN(num_of_entries); if (num_of_entries > 0) { for (VarNameValMap::const_iterator iter = map_.begin(); (iter != map_.end()); iter++) { const ObString &key = iter->first; const ObSessionVariable &value = iter->second; OB_UNIS_ADD_LEN(key); OB_UNIS_ADD_LEN(value.meta_); OB_UNIS_ADD_LEN(value.value_); } } return len; }