| // Licensed to the Apache Software Foundation (ASF) under one | |
| // or more contributor license agreements. See the NOTICE file | |
| // distributed with this work for additional information | |
| // regarding copyright ownership. The ASF licenses this file | |
| // to you 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. See the License for the | |
| // specific language governing permissions and limitations | |
| // under the License. | |
| #include "runtime/cache/result_node.h" | |
| #include <gen_cpp/internal_service.pb.h> | |
| #include <gen_cpp/types.pb.h> | |
| #include <glog/logging.h> | |
| #include <iostream> | |
| #include <limits> | |
| #include <utility> | |
| #include "common/config.h" | |
| #include "runtime/cache/cache_utils.h" | |
| #include "storage/olap_define.h" | |
| namespace doris { | |
| bool compare_partition(const PartitionRowBatch* left_node, const PartitionRowBatch* right_node) { | |
| return left_node->get_partition_key() < right_node->get_partition_key(); | |
| } | |
| //return new batch size,only include the size of PRowBatch | |
| void PartitionRowBatch::set_row_batch(const PCacheValue& value) { | |
| if (_cache_value != nullptr && !check_newer(value.param())) { | |
| LOG(WARNING) << "set old version data, cache ver:" << _cache_value->param().last_version() | |
| << ",cache time:" << _cache_value->param().last_version_time() | |
| << ", setdata ver:" << value.param().last_version() | |
| << ",setdata time:" << value.param().last_version_time(); | |
| return; | |
| } | |
| SAFE_DELETE(_cache_value); | |
| _cache_value = new PCacheValue(value); | |
| _data_size += _cache_value->data_size(); | |
| _cache_stat.update(); | |
| LOG(INFO) << "finish set row batch, row num:" << _cache_value->rows_size() | |
| << ", data size:" << _data_size; | |
| } | |
| bool PartitionRowBatch::is_hit_cache(const PCacheParam& param) { | |
| if (!check_match(param)) { | |
| return false; | |
| } | |
| _cache_stat.query(); | |
| return true; | |
| } | |
| void PartitionRowBatch::clear() { | |
| LOG(INFO) << "clear partition rowbatch."; | |
| SAFE_DELETE(_cache_value); | |
| _partition_key = 0; | |
| _data_size = 0; | |
| _cache_stat.init(); | |
| } | |
| /** | |
| * Update partition cache data, find RowBatch from partition map by partition key, | |
| * the partition rowbatch are stored in the order of partition keys | |
| */ | |
| PCacheStatus ResultNode::update_partition(const PUpdateCacheRequest* request, | |
| bool& is_update_firstkey) { | |
| is_update_firstkey = false; | |
| if (_sql_key != request->sql_key()) { | |
| LOG(INFO) << "no match sql_key " << request->sql_key().hi() << request->sql_key().lo(); | |
| return PCacheStatus::PARAM_ERROR; | |
| } | |
| if (request->values_size() > config::query_cache_max_partition_count) { | |
| LOG(WARNING) << "too many partitions size:" << request->values_size(); | |
| return PCacheStatus::PARAM_ERROR; | |
| } | |
| //Only one thread per SQL key can update the cache | |
| CacheWriteLock write_lock(_node_mtx); | |
| if (request->cache_type() == CacheType::SQL_CACHE) { | |
| return update_sql_cache(request, is_update_firstkey); | |
| } else { | |
| return update_partition_cache(request, is_update_firstkey); | |
| } | |
| } | |
| PCacheStatus ResultNode::update_sql_cache(const PUpdateCacheRequest* request, | |
| bool& is_update_firstkey) { | |
| PartitionRowBatch* partition = nullptr; | |
| if (request->values_size() > 1) { | |
| return PCacheStatus::PARAM_ERROR; | |
| } | |
| is_update_firstkey = true; | |
| const PCacheValue& value = request->values(0); | |
| PartitionKey partition_key = value.param().partition_key(); | |
| // no cache exist, create new cache node | |
| if (_partition_map.size() == 0) { | |
| partition = new PartitionRowBatch(partition_key); | |
| partition->set_row_batch(value); | |
| _partition_map[partition_key] = partition; | |
| _partition_list.push_back(partition); | |
| } else { | |
| // compatible with previous version | |
| for (auto it = _partition_list.begin(); it != _partition_list.end(); it++) { | |
| _data_size -= (*it)->get_data_size(); | |
| } | |
| // clear old cache, and create new cache node | |
| for (auto it = _partition_list.begin(); it != _partition_list.end();) { | |
| (*it)->clear(); | |
| SAFE_DELETE(*it); | |
| it = _partition_list.erase(it); | |
| } | |
| _partition_map.clear(); | |
| partition = new PartitionRowBatch(partition_key); | |
| partition->set_row_batch(value); | |
| _partition_map[partition_key] = partition; | |
| _partition_list.push_back(partition); | |
| } | |
| _data_size += partition->get_data_size(); | |
| VLOG(1) << "finish update sql cache batches:" << _partition_list.size(); | |
| return PCacheStatus::CACHE_OK; | |
| } | |
| PCacheStatus ResultNode::update_partition_cache(const PUpdateCacheRequest* request, | |
| bool& is_update_firstkey) { | |
| PartitionKey first_key = std::numeric_limits<long>::max(); | |
| if (_partition_list.size() == 0) { | |
| is_update_firstkey = true; | |
| } else { | |
| first_key = (*(_partition_list.begin()))->get_partition_key(); | |
| } | |
| PartitionRowBatch* partition = nullptr; | |
| for (int i = 0; i < request->values_size(); i++) { | |
| const PCacheValue& value = request->values(i); | |
| PartitionKey partition_key = value.param().partition_key(); | |
| if (!is_update_firstkey && partition_key <= first_key) { | |
| is_update_firstkey = true; | |
| } | |
| auto it = _partition_map.find(partition_key); | |
| if (it == _partition_map.end()) { | |
| partition = new PartitionRowBatch(partition_key); | |
| partition->set_row_batch(value); | |
| _partition_map[partition_key] = partition; | |
| _partition_list.push_back(partition); | |
| #ifdef PARTITION_CACHE_DEV | |
| LOG(INFO) << "add index:" << i << ", pkey:" << partition->get_partition_key() | |
| << ", list size:" << _partition_list.size() | |
| << ", map size:" << _partition_map.size(); | |
| #endif | |
| } else { | |
| partition = it->second; | |
| _data_size -= partition->get_data_size(); | |
| partition->set_row_batch(value); | |
| #ifdef PARTITION_CACHE_DEV | |
| LOG(INFO) << "update index:" << i << ", pkey:" << partition->get_partition_key() | |
| << ", list size:" << _partition_list.size() | |
| << ", map size:" << _partition_map.size(); | |
| #endif | |
| } | |
| _data_size += partition->get_data_size(); | |
| } | |
| _partition_list.sort(compare_partition); | |
| VLOG(1) << "finish update partition cache batches:" << _partition_list.size(); | |
| while (config::query_cache_max_partition_count > 0 && | |
| _partition_list.size() > config::query_cache_max_partition_count) { | |
| if (prune_first() == 0) { | |
| break; | |
| } | |
| } | |
| return PCacheStatus::CACHE_OK; | |
| } | |
| /** | |
| * Only the range query of the key of the partition is supported, and the separated partition key query is not supported. | |
| * Because a query can only be divided into two parts, part1 get data from cache, part2 fetch_data by scan node from BE. | |
| * Partition cache : 20191211-20191215 | |
| * Hit cache parameter : [20191211 - 20191215], [20191212 - 20191214], [20191212 - 20191216],[20191210 - 20191215] | |
| * Miss cache parameter: [20191210 - 20191216] | |
| */ | |
| PCacheStatus ResultNode::fetch_partition(const PFetchCacheRequest* request, | |
| PartitionRowBatchList& row_batch_list, | |
| bool& is_hit_firstkey) { | |
| is_hit_firstkey = false; | |
| if (request->params_size() == 0) { | |
| return PCacheStatus::PARAM_ERROR; | |
| } | |
| CacheReadLock read_lock(_node_mtx); | |
| if (_partition_list.size() == 0) { | |
| return PCacheStatus::NO_PARTITION_KEY; | |
| } | |
| if (request->params(0).partition_key() > (*_partition_list.rbegin())->get_partition_key() || | |
| request->params(request->params_size() - 1).partition_key() < | |
| (*_partition_list.begin())->get_partition_key()) { | |
| return PCacheStatus::NO_PARTITION_KEY; | |
| } | |
| bool find = false; | |
| int begin_idx = -1, end_idx = -1, param_idx = 0; | |
| auto begin_it = _partition_list.end(); | |
| auto end_it = _partition_list.end(); | |
| auto part_it = _partition_list.begin(); | |
| PCacheStatus status = PCacheStatus::CACHE_OK; | |
| while (param_idx < request->params_size() && part_it != _partition_list.end()) { | |
| #ifdef PARTITION_CACHE_DEV | |
| LOG(INFO) << "Param index : " << param_idx | |
| << ", param part Key : " << request->params(param_idx).partition_key() | |
| << ", batch part key : " << (*part_it)->get_partition_key(); | |
| #endif | |
| if (!find) { | |
| while (part_it != _partition_list.end() && | |
| request->params(param_idx).partition_key() > (*part_it)->get_partition_key()) { | |
| part_it++; | |
| } | |
| if (part_it != _partition_list.end()) { | |
| while (param_idx < request->params_size() && | |
| request->params(param_idx).partition_key() < | |
| (*part_it)->get_partition_key()) { | |
| param_idx++; | |
| } | |
| if (param_idx < request->params_size()) { | |
| if (request->params(param_idx).partition_key() == | |
| (*part_it)->get_partition_key()) { | |
| find = true; | |
| } | |
| } | |
| } | |
| } | |
| if (find) { | |
| #ifdef PARTITION_CACHE_DEV | |
| LOG(INFO) << "Find! Param index : " << param_idx | |
| << ", param part Key : " << request->params(param_idx).partition_key() | |
| << ", batch part key : " << (*part_it)->get_partition_key() | |
| << ", param part version : " << request->params(param_idx).last_version() | |
| << ", batch part version : " | |
| << (*part_it)->get_value()->param().last_version() | |
| << ", param part version time : " | |
| << request->params(param_idx).last_version_time() | |
| << ", batch part version time : " | |
| << (*part_it)->get_value()->param().last_version_time(); | |
| #endif | |
| if ((*part_it)->is_hit_cache(request->params(param_idx))) { | |
| if (begin_idx < 0) { | |
| begin_idx = param_idx; | |
| begin_it = part_it; | |
| } | |
| end_idx = param_idx; | |
| end_it = part_it; | |
| param_idx++; | |
| part_it++; | |
| find = false; | |
| } else { | |
| status = PCacheStatus::DATA_OVERDUE; | |
| break; | |
| } | |
| } | |
| } | |
| if (begin_it == _partition_list.end() && end_it == _partition_list.end()) { | |
| return status; | |
| } | |
| //[20191210 - 20191216] hit partition range [20191212-20191214],the sql will be splited to 3 part! | |
| if (begin_idx != 0 && end_idx != request->params_size() - 1) { | |
| return PCacheStatus::INVALID_KEY_RANGE; | |
| } | |
| if (begin_it == _partition_list.begin()) { | |
| is_hit_firstkey = true; | |
| } | |
| while (true) { | |
| row_batch_list.push_back(*begin_it); | |
| if (begin_it == end_it) { | |
| break; | |
| } | |
| begin_it++; | |
| } | |
| return status; | |
| } | |
| /* | |
| * prune first partition result | |
| */ | |
| size_t ResultNode::prune_first() { | |
| if (_partition_list.size() == 0) { | |
| return 0; | |
| } | |
| PartitionRowBatch* part_node = *_partition_list.begin(); | |
| size_t prune_size = part_node->get_data_size(); | |
| _partition_list.erase(_partition_list.begin()); | |
| _partition_map.erase(part_node->get_partition_key()); | |
| part_node->clear(); | |
| SAFE_DELETE(part_node); | |
| _data_size -= prune_size; | |
| return prune_size; | |
| } | |
| void ResultNode::clear() { | |
| CacheWriteLock write_lock(_node_mtx); | |
| LOG(INFO) << "clear result node:" << _sql_key; | |
| _sql_key.hi = 0; | |
| _sql_key.lo = 0; | |
| for (auto it = _partition_list.begin(); it != _partition_list.end();) { | |
| (*it)->clear(); | |
| SAFE_DELETE(*it); | |
| it = _partition_list.erase(it); | |
| } | |
| _data_size = 0; | |
| } | |
| void ResultNode::append(ResultNode* tail) { | |
| _prev = tail; | |
| if (tail) tail->set_next(this); | |
| } | |
| void ResultNode::unlink() { | |
| if (_next) { | |
| _next->set_prev(_prev); | |
| } | |
| if (_prev) { | |
| _prev->set_next(_next); | |
| } | |
| _next = nullptr; | |
| _prev = nullptr; | |
| } | |
| } // namespace doris |