// 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 "common/metrics/doris_metrics.h"

#include "runtime/cache/cache_utils.h"

#include "storage/olap_define.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