be/src/runtime/cache/result_node.h

// 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.

#pragma once

#include <gen_cpp/internal_service.pb.h>

#include <cstdio>
#include <list>
#include <shared_mutex>
#include <unordered_map>

#include "runtime/cache/cache_utils.h"
#include "util/uid_util.h"

namespace doris {

/**
* Cache one partition data, request param must match version and time of cache
*/
class PartitionRowBatch {
public:
    PartitionRowBatch(int64_t partition_key)
            : _partition_key(partition_key), _cache_value(nullptr), _data_size(0) {}

    ~PartitionRowBatch() {}

    void set_row_batch(const PCacheValue& value);
    bool is_hit_cache(const PCacheParam& param);
    void clear();

    int64_t get_partition_key() const { return _partition_key; }

    PCacheValue* get_value() { return _cache_value; }

    size_t get_data_size() { return _data_size; }

    const CacheStat* get_stat() const { return &_cache_stat; }

private:
    bool check_match(const PCacheParam& req_param) {
        if (req_param.partition_key() != _partition_key) {
            return false;
        }
        if (req_param.last_version() > _cache_value->param().last_version()) {
            return false;
        }
        if (req_param.last_version_time() > _cache_value->param().last_version_time()) {
            return false;
        }
        if (req_param.partition_num() != _cache_value->param().partition_num()) {
            return false;
        }
        return true;
    }

    bool check_newer(const PCacheParam& up_param) {
        //for init data of sql cache
        if (up_param.last_version() == 0 || up_param.last_version_time() == 0) {
            return true;
        }
        if (up_param.last_version_time() > _cache_value->param().last_version_time()) {
            return true;
        }
        if (up_param.last_version_time() == _cache_value->param().last_version_time() &&
            up_param.partition_num() != _cache_value->param().partition_num()) {
            return true;
        }
        if (up_param.last_version() > _cache_value->param().last_version()) {
            return true;
        }
        if (up_param.last_version() == _cache_value->param().last_version() &&
            up_param.partition_num() != _cache_value->param().partition_num()) {
            return true;
        }
        return false;
    }

private:
    int64_t _partition_key;
    PCacheValue* _cache_value = nullptr;
    size_t _data_size;
    CacheStat _cache_stat;
};

typedef int64_t PartitionKey;
typedef std::list<PartitionRowBatch*> PartitionRowBatchList;
typedef std::unordered_map<PartitionKey, PartitionRowBatch*> PartitionRowBatchMap;

/**
* Cache the result of one SQL, include many partition rowsets.
* Sql Cache: The partition ID comes from the partition last updated.
* Partition Cache: The partition ID comes from the partition scanned by query.
* The above two modes use the same cache structure.
*/
class ResultNode {
public:
    ResultNode() : _sql_key(0, 0), _prev(nullptr), _next(nullptr), _data_size(0) {}

    ResultNode(const UniqueId& sql_key)
            : _sql_key(sql_key), _prev(nullptr), _next(nullptr), _data_size(0) {}

    virtual ~ResultNode() {}

    PCacheStatus update_partition(const PUpdateCacheRequest* request, bool& is_update_firstkey);
    PCacheStatus fetch_partition(const PFetchCacheRequest* request,
                                 PartitionRowBatchList& rowBatchList, bool& is_hit_firstkey);
    PCacheStatus update_sql_cache(const PUpdateCacheRequest* request, bool& is_update_firstkey);
    PCacheStatus update_partition_cache(const PUpdateCacheRequest* request,
                                        bool& is_update_firstkey);

    size_t prune_first();
    void clear();

    ResultNode* get_prev() { return _prev; }

    void set_prev(ResultNode* prev) { _prev = prev; }

    ResultNode* get_next() { return _next; }

    void set_next(ResultNode* next) { _next = next; }

    void append(ResultNode* tail);

    void unlink();

    size_t get_partition_count() const { return _partition_list.size(); }

    size_t get_data_size() const { return _data_size; }

    UniqueId get_sql_key() { return _sql_key; }

    bool sql_key_null() { return _sql_key.hi == 0 && _sql_key.lo == 0; }

    void set_sql_key(const UniqueId& sql_key) { _sql_key = sql_key; }

    long first_partition_last_time() const {
        if (_partition_list.size() == 0) {
            return 0;
        }
        const PartitionRowBatch* first = *(_partition_list.begin());
        return first->get_stat()->last_read_time;
    }

    const CacheStat* get_first_stat() const {
        if (_partition_list.size() == 0) {
            return nullptr;
        }
        return (*(_partition_list.begin()))->get_stat();
    }

    const CacheStat* get_last_stat() const {
        if (_partition_list.size() == 0) {
            return nullptr;
        }
        return (*(_partition_list.end()--))->get_stat();
    }

private:
    mutable std::shared_mutex _node_mtx;
    UniqueId _sql_key;
    ResultNode* _prev = nullptr;
    ResultNode* _next = nullptr;
    size_t _data_size;
    PartitionRowBatchList _partition_list;
    PartitionRowBatchMap _partition_map;
};

} // namespace doris