be/src/runtime/cache/result_cache.cpp - doris - Git at Google

 // 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_cache.h"

 #include <gen_cpp/internal_service.pb.h>
 #include <glog/logging.h>

 #include <iostream>
 #include <list>
 #include <utility>

 #include "olap/olap_define.h"
 #include "runtime/cache/cache_utils.h"
 #include "util/doris_metrics.h"

 namespace doris {

 /**
 * Remove the tail node of link
 */
 ResultNode* ResultNodeList::pop() {
     remove(_head);
     return _head;
 }

 void ResultNodeList::remove(ResultNode* node) {
     if (!node) return;
     if (node == _head) _head = node->get_next();
     if (node == _tail) _tail = node->get_prev();
     node->unlink();
     _node_count--;
 }

 void ResultNodeList::push_back(ResultNode* node) {
     if (!node) return;
     if (!_head) _head = node;
     node->append(_tail);
     _tail = node;
     _node_count++;
 }

 void ResultNodeList::move_tail(ResultNode* node) {
     if (!node || node == _tail) return;
     if (!_head)
         _head = node;
     else if (node == _head)
         _head = node->get_next();
     node->unlink();
     node->append(_tail);
     _tail = node;
 }

 void ResultNodeList::delete_node(ResultNode** node) {
     (*node)->clear();
     SAFE_DELETE(*node);
 }

 void ResultNodeList::clear() {
     LOG(INFO) << "clear result node list.";
     while (_head) {
         ResultNode* tmp_node = _head->get_next();
         _head->clear();
         SAFE_DELETE(_head);
         _head = tmp_node;
     }
     _node_count = 0;
 }
 /**
  * Find the node and update partition data
  * New node, the node updated in the first partition will move to the tail of the list
  */
 void ResultCache::update(const PUpdateCacheRequest* request, PCacheResponse* response) {
     ResultNode* node;
     PCacheStatus status;
     bool update_first = false;
     UniqueId sql_key = request->sql_key();
     LOG(INFO) << "update cache, sql key:" << sql_key;

     CacheWriteLock write_lock(_cache_mtx);
     auto it = _node_map.find(sql_key);
     if (it != _node_map.end()) {
         node = it->second;
         _cache_size -= node->get_data_size();
         _partition_count -= node->get_partition_count();
         status = node->update_partition(request, update_first);
     } else {
         node = _node_list.new_node(sql_key);
         status = node->update_partition(request, update_first);
         _node_list.push_back(node);
         _node_map[sql_key] = node;
         _node_count += 1;
     }
     if (update_first) {
         _node_list.move_tail(node);
     }
     _cache_size += node->get_data_size();
     _partition_count += node->get_partition_count();
     response->set_status(status);

     prune();
     update_monitor();
 }

 /**
  * Fetch cache through sql key, partition key, version and time
  */
 void ResultCache::fetch(const PFetchCacheRequest* request, PFetchCacheResult* result) {
     bool hit_first = false;
     ResultNodeMap::iterator node_it;
     const UniqueId sql_key = request->sql_key();
     LOG(INFO) << "fetch cache, sql key:" << sql_key;
     {
         CacheReadLock read_lock(_cache_mtx);
         node_it = _node_map.find(sql_key);
         if (node_it == _node_map.end()) {
             result->set_status(PCacheStatus::NO_SQL_KEY);
             LOG(INFO) << "no such sql key:" << sql_key;
             return;
         }
         ResultNode* node = node_it->second;
         PartitionRowBatchList part_rowbatch_list;
         PCacheStatus status = node->fetch_partition(request, part_rowbatch_list, hit_first);

         for (auto part_it = part_rowbatch_list.begin(); part_it != part_rowbatch_list.end();
              part_it++) {
             PCacheValue* srcValue = (*part_it)->get_value();
             if (srcValue != nullptr) {
                 PCacheValue* value = result->add_values();
                 value->CopyFrom(*srcValue);
                 LOG(INFO) << "fetch cache partition key:" << srcValue->param().partition_key();
             } else {
                 LOG(WARNING) << "prowbatch of cache is null";
                 status = PCacheStatus::EMPTY_DATA;
                 break;
             }
         }
         if (status == PCacheStatus::CACHE_OK && part_rowbatch_list.empty()) {
             status = PCacheStatus::EMPTY_DATA;
         }
         result->set_status(status);
     }

     if (hit_first) {
         CacheWriteLock write_lock(_cache_mtx);
         _node_list.move_tail(node_it->second);
     }
 }

 bool ResultCache::contains(const UniqueId& sql_key) {
     CacheReadLock read_lock(_cache_mtx);
     return _node_map.find(sql_key) != _node_map.end();
 }

 /**
  * enum PClearType {
  *   CLEAR_ALL = 0,
  *   PRUNE_CACHE = 1,
  *   CLEAR_BEFORE_TIME = 2,
  *   CLEAR_SQL_KEY = 3
  * };
  */
 void ResultCache::clear(const PClearCacheRequest* request, PCacheResponse* response) {
     LOG(INFO) << "clear cache type" << request->clear_type()
               << ", node size:" << _node_list.get_node_count() << ", map size:" << _node_map.size();
     CacheWriteLock write_lock(_cache_mtx);
     //0 clear, 1 prune, 2 before_time,3 sql_key
     switch (request->clear_type()) {
     case PClearType::CLEAR_ALL:
         _node_list.clear();
         _node_map.clear();
         _cache_size = 0;
         _node_count = 0;
         _partition_count = 0;
         break;
     case PClearType::PRUNE_CACHE:
         prune();
         break;
     default:
         break;
     }
     update_monitor();
     response->set_status(PCacheStatus::CACHE_OK);
 }

 //private method
 ResultNode* find_min_time_node(ResultNode* result_node) {
     if (result_node->get_prev()) {
         if (result_node->get_prev()->first_partition_last_time() <=
             result_node->first_partition_last_time()) {
             return result_node->get_prev();
         }
     }

     if (result_node->get_next()) {
         if (result_node->get_next()->first_partition_last_time() <
             result_node->first_partition_last_time()) {
             return result_node->get_next();
         }
     }
     return result_node;
 }

 /*
 * Two-dimensional array, prune the min last_read_time PartitionRowBatch.
 * The following example is the last read time array.
 * 1 and 2 is the read time, nodes with pruning read time < 3
 * Before:
 *   1,2         //_head ResultNode*
 *   1,2,3,4,5
 *   2,4,3,6,8
 *   5,7,9,11,13 //_tail ResultNode*
 * After:
 *   4,5         //_head
 *   4,3,6,8
 *   5,7,9,11,13 //_tail
 */
 void ResultCache::prune() {
     if (_cache_size <= (_max_size + _elasticity_size)) {
         return;
     }
     LOG(INFO) << "begin prune cache, cache_size : " << _cache_size << ", max_size : " << _max_size
               << ", elasticity_size : " << _elasticity_size;
     ResultNode* result_node = _node_list.get_head();
     while (_cache_size > _max_size) {
         if (result_node == nullptr) {
             break;
         }
         result_node = find_min_time_node(result_node);
         _cache_size -= result_node->prune_first();
         if (result_node->get_data_size() == 0) {
             ResultNode* next_node;
             if (result_node->get_next()) {
                 next_node = result_node->get_next();
             } else if (result_node->get_prev()) {
                 next_node = result_node->get_prev();
             } else {
                 next_node = _node_list.get_head();
             }
             remove(result_node);
             result_node = next_node;
         }
     }
     LOG(INFO) << "finish prune, cache_size : " << _cache_size;
     _node_count = _node_map.size();
     _cache_size = 0;
     _partition_count = 0;
     for (auto node_it = _node_map.begin(); node_it != _node_map.end(); node_it++) {
         _partition_count += node_it->second->get_partition_count();
         _cache_size += node_it->second->get_data_size();
     }
 }

 void ResultCache::remove(ResultNode* result_node) {
     auto node_it = _node_map.find(result_node->get_sql_key());
     if (node_it != _node_map.end()) {
         _node_map.erase(node_it);
         _node_list.remove(result_node);
         _node_list.delete_node(&result_node);
     }
 }

 void ResultCache::update_monitor() {
     DorisMetrics::instance()->query_cache_memory_total_byte->set_value(_cache_size);
     DorisMetrics::instance()->query_cache_sql_total_count->set_value(_node_count);
     DorisMetrics::instance()->query_cache_partition_total_count->set_value(_partition_count);
 }

 } // namespace doris
	// 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_cache.h"

	#include <gen_cpp/internal_service.pb.h>
	#include <glog/logging.h>

	#include <iostream>
	#include <list>
	#include <utility>

	#include "olap/olap_define.h"
	#include "runtime/cache/cache_utils.h"
	#include "util/doris_metrics.h"

	namespace doris {

	/**
	* Remove the tail node of link
	*/
	ResultNode* ResultNodeList::pop() {
	remove(_head);
	return _head;
	}

	void ResultNodeList::remove(ResultNode* node) {
	if (!node) return;
	if (node == _head) _head = node->get_next();
	if (node == _tail) _tail = node->get_prev();
	node->unlink();
	_node_count--;
	}

	void ResultNodeList::push_back(ResultNode* node) {
	if (!node) return;
	if (!_head) _head = node;
	node->append(_tail);
	_tail = node;
	_node_count++;
	}

	void ResultNodeList::move_tail(ResultNode* node) {
	if (!node \|\| node == _tail) return;
	if (!_head)
	_head = node;
	else if (node == _head)
	_head = node->get_next();
	node->unlink();
	node->append(_tail);
	_tail = node;
	}

	void ResultNodeList::delete_node(ResultNode** node) {
	(*node)->clear();
	SAFE_DELETE(*node);
	}

	void ResultNodeList::clear() {
	LOG(INFO) << "clear result node list.";
	while (_head) {
	ResultNode* tmp_node = _head->get_next();
	_head->clear();
	SAFE_DELETE(_head);
	_head = tmp_node;
	}
	_node_count = 0;
	}
	/**
	* Find the node and update partition data
	* New node, the node updated in the first partition will move to the tail of the list
	*/
	void ResultCache::update(const PUpdateCacheRequest* request, PCacheResponse* response) {
	ResultNode* node;
	PCacheStatus status;
	bool update_first = false;
	UniqueId sql_key = request->sql_key();
	LOG(INFO) << "update cache, sql key:" << sql_key;

	CacheWriteLock write_lock(_cache_mtx);
	auto it = _node_map.find(sql_key);
	if (it != _node_map.end()) {
	node = it->second;
	_cache_size -= node->get_data_size();
	_partition_count -= node->get_partition_count();
	status = node->update_partition(request, update_first);
	} else {
	node = _node_list.new_node(sql_key);
	status = node->update_partition(request, update_first);
	_node_list.push_back(node);
	_node_map[sql_key] = node;
	_node_count += 1;
	}
	if (update_first) {
	_node_list.move_tail(node);
	}
	_cache_size += node->get_data_size();
	_partition_count += node->get_partition_count();
	response->set_status(status);

	prune();
	update_monitor();
	}

	/**
	* Fetch cache through sql key, partition key, version and time
	*/
	void ResultCache::fetch(const PFetchCacheRequest* request, PFetchCacheResult* result) {
	bool hit_first = false;
	ResultNodeMap::iterator node_it;
	const UniqueId sql_key = request->sql_key();
	LOG(INFO) << "fetch cache, sql key:" << sql_key;
	{
	CacheReadLock read_lock(_cache_mtx);
	node_it = _node_map.find(sql_key);
	if (node_it == _node_map.end()) {
	result->set_status(PCacheStatus::NO_SQL_KEY);
	LOG(INFO) << "no such sql key:" << sql_key;
	return;
	}
	ResultNode* node = node_it->second;
	PartitionRowBatchList part_rowbatch_list;
	PCacheStatus status = node->fetch_partition(request, part_rowbatch_list, hit_first);

	for (auto part_it = part_rowbatch_list.begin(); part_it != part_rowbatch_list.end();
	part_it++) {
	PCacheValue* srcValue = (*part_it)->get_value();
	if (srcValue != nullptr) {
	PCacheValue* value = result->add_values();
	value->CopyFrom(*srcValue);
	LOG(INFO) << "fetch cache partition key:" << srcValue->param().partition_key();
	} else {
	LOG(WARNING) << "prowbatch of cache is null";
	status = PCacheStatus::EMPTY_DATA;
	break;
	}
	}
	if (status == PCacheStatus::CACHE_OK && part_rowbatch_list.empty()) {
	status = PCacheStatus::EMPTY_DATA;
	}
	result->set_status(status);
	}

	if (hit_first) {
	CacheWriteLock write_lock(_cache_mtx);
	_node_list.move_tail(node_it->second);
	}
	}

	bool ResultCache::contains(const UniqueId& sql_key) {
	CacheReadLock read_lock(_cache_mtx);
	return _node_map.find(sql_key) != _node_map.end();
	}

	/**
	* enum PClearType {
	* CLEAR_ALL = 0,
	* PRUNE_CACHE = 1,
	* CLEAR_BEFORE_TIME = 2,
	* CLEAR_SQL_KEY = 3
	* };
	*/
	void ResultCache::clear(const PClearCacheRequest* request, PCacheResponse* response) {
	LOG(INFO) << "clear cache type" << request->clear_type()
	<< ", node size:" << _node_list.get_node_count() << ", map size:" << _node_map.size();
	CacheWriteLock write_lock(_cache_mtx);
	//0 clear, 1 prune, 2 before_time,3 sql_key
	switch (request->clear_type()) {
	case PClearType::CLEAR_ALL:
	_node_list.clear();
	_node_map.clear();
	_cache_size = 0;
	_node_count = 0;
	_partition_count = 0;
	break;
	case PClearType::PRUNE_CACHE:
	prune();
	break;
	default:
	break;
	}
	update_monitor();
	response->set_status(PCacheStatus::CACHE_OK);
	}

	//private method
	ResultNode* find_min_time_node(ResultNode* result_node) {
	if (result_node->get_prev()) {
	if (result_node->get_prev()->first_partition_last_time() <=
	result_node->first_partition_last_time()) {
	return result_node->get_prev();
	}
	}

	if (result_node->get_next()) {
	if (result_node->get_next()->first_partition_last_time() <
	result_node->first_partition_last_time()) {
	return result_node->get_next();
	}
	}
	return result_node;
	}

	/*
	* Two-dimensional array, prune the min last_read_time PartitionRowBatch.
	* The following example is the last read time array.
	* 1 and 2 is the read time, nodes with pruning read time < 3
	* Before:
	* 1,2 //_head ResultNode*
	* 1,2,3,4,5
	* 2,4,3,6,8
	* 5,7,9,11,13 //_tail ResultNode*
	* After:
	* 4,5 //_head
	* 4,3,6,8
	* 5,7,9,11,13 //_tail
	*/
	void ResultCache::prune() {
	if (_cache_size <= (_max_size + _elasticity_size)) {
	return;
	}
	LOG(INFO) << "begin prune cache, cache_size : " << _cache_size << ", max_size : " << _max_size
	<< ", elasticity_size : " << _elasticity_size;
	ResultNode* result_node = _node_list.get_head();
	while (_cache_size > _max_size) {
	if (result_node == nullptr) {
	break;
	}
	result_node = find_min_time_node(result_node);
	_cache_size -= result_node->prune_first();
	if (result_node->get_data_size() == 0) {
	ResultNode* next_node;
	if (result_node->get_next()) {
	next_node = result_node->get_next();
	} else if (result_node->get_prev()) {
	next_node = result_node->get_prev();
	} else {
	next_node = _node_list.get_head();
	}
	remove(result_node);
	result_node = next_node;
	}
	}
	LOG(INFO) << "finish prune, cache_size : " << _cache_size;
	_node_count = _node_map.size();
	_cache_size = 0;
	_partition_count = 0;
	for (auto node_it = _node_map.begin(); node_it != _node_map.end(); node_it++) {
	_partition_count += node_it->second->get_partition_count();
	_cache_size += node_it->second->get_data_size();
	}
	}

	void ResultCache::remove(ResultNode* result_node) {
	auto node_it = _node_map.find(result_node->get_sql_key());
	if (node_it != _node_map.end()) {
	_node_map.erase(node_it);
	_node_list.remove(result_node);
	_node_list.delete_node(&result_node);
	}
	}

	void ResultCache::update_monitor() {
	DorisMetrics::instance()->query_cache_memory_total_byte->set_value(_cache_size);
	DorisMetrics::instance()->query_cache_sql_total_count->set_value(_node_count);
	DorisMetrics::instance()->query_cache_partition_total_count->set_value(_partition_count);
	}

	} // namespace doris