morningman commented on a change in pull request #5793:
URL: https://github.com/apache/incubator-doris/pull/5793#discussion_r637651299
##########
File path: gensrc/proto/internal_service.proto
##########
@@ -184,6 +189,7 @@ message PCacheResponse {
message PUpdateCacheRequest{
required PUniqueId sql_key = 1;
repeated PCacheValue values = 2;
+ required CacheType cache_type = 3;
Review comment:
use optional to make it compatible with old version?
##########
File path: be/src/runtime/cache/result_node.cpp
##########
@@ -77,7 +77,48 @@ PCacheStatus ResultNode::update_partition(const
PUpdateCacheRequest* request,
//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 if (request->cache_type() == CacheType::PARTITION_CACHE) {
+ return update_partition_cache(request, is_update_firstkey);
+ } else {
+ return PCacheStatus::PARAM_ERROR;
+ }
+}
+
+PCacheStatus ResultNode::update_sql_cache(const PUpdateCacheRequest *request,
bool &is_update_firstkey) {
+ PartitionRowBatch* partition = NULL;
+ 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
+ _partition_map.clear();
Review comment:
The _partition_map save the pointer of `PartitionRowBatch`, so call
`clear()` will not release to memory used by `PartitionRowBatch`.
And also I checked the origin code, it seems that the `PartitionRowBatch` in
`_partition_map` is never released?
##########
File path: be/src/runtime/cache/result_node.cpp
##########
@@ -115,7 +156,7 @@ PCacheStatus ResultNode::update_partition(const
PUpdateCacheRequest* request,
_data_size += partition->get_data_size();
}
_partition_list.sort(compare_partition);
- LOG(INFO) << "finish update batches:" << _partition_list.size();
+ LOG(INFO) << "finish update partition cache batches:" <<
_partition_list.size();
Review comment:
using debug level maybe better.
##########
File path: be/src/runtime/cache/result_node.cpp
##########
@@ -77,7 +77,48 @@ PCacheStatus ResultNode::update_partition(const
PUpdateCacheRequest* request,
//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 if (request->cache_type() == CacheType::PARTITION_CACHE) {
+ return update_partition_cache(request, is_update_firstkey);
+ } else {
+ return PCacheStatus::PARAM_ERROR;
+ }
+}
+
+PCacheStatus ResultNode::update_sql_cache(const PUpdateCacheRequest *request,
bool &is_update_firstkey) {
+ PartitionRowBatch* partition = NULL;
+ 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
+ _partition_map.clear();
+ _partition_list.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();
+ LOG(INFO) << "finish update sql cache batches:" << _partition_list.size();
Review comment:
Using debug level.
##########
File path: be/src/runtime/cache/result_node.cpp
##########
@@ -77,7 +77,48 @@ PCacheStatus ResultNode::update_partition(const
PUpdateCacheRequest* request,
//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 if (request->cache_type() == CacheType::PARTITION_CACHE) {
+ return update_partition_cache(request, is_update_firstkey);
+ } else {
+ return PCacheStatus::PARAM_ERROR;
Review comment:
we need to make it compatible with old version
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
