be/src/storage/segment/condition_cache.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 <butil/macros.h>
#include <glog/logging.h>
#include <stddef.h>
#include <stdint.h>

#include <atomic>
#include <memory>
#include <roaring/roaring.hh>
#include <string>
#include <vector>

#include "common/config.h"
#include "common/status.h"
#include "io/fs/file_system.h"
#include "io/fs/path.h"
#include "runtime/exec_env.h"
#include "runtime/memory/lru_cache_policy.h"
#include "runtime/memory/mem_tracker.h"
#include "util/lru_cache.h"
#include "util/slice.h"
#include "util/time.h"

namespace doris {

// Context passed from scan/table-reader layers to physical readers for condition cache
// integration. On MISS, readers set filter_result[granule] to true when row-level predicates keep
// at least one row in that granule. On HIT, readers skip granules whose cached bit is false.
struct ConditionCacheContext {
    bool is_hit = false;
    std::shared_ptr<std::vector<bool>> filter_result; // per-granule: true = has surviving rows
    int64_t base_granule = 0;                         // global granule index of filter_result[0]
    size_t num_granules = 0; // authoritative bitmap length; excludes allocation-only guard bits
    static constexpr int GRANULE_SIZE = 2048;
};

namespace segment_v2 {

class ConditionCacheHandle;

class ConditionCache : public LRUCachePolicy {
public:
    using LRUCachePolicy::insert;

    // The cache key or segment lru cache
    struct CacheKey {
        CacheKey(RowsetId rowset_id_, int64_t segment_id_, uint64_t digest_)
                : rowset_id(rowset_id_), segment_id(segment_id_), digest(digest_) {}
        RowsetId rowset_id;
        int64_t segment_id;
        uint64_t digest;

        // Encode to a flat binary which can be used as LRUCache's key
        [[nodiscard]] std::string encode() const {
            char buf[16];
            memcpy(buf, &segment_id, 8);
            memcpy(buf + 8, &digest, 8);

            return rowset_id.to_string() + std::string(buf, 16);
        }
    };

    class CacheValue : public LRUCacheValueBase {
    public:
        std::shared_ptr<std::vector<bool>> filter_result;
        // The bitmap coordinate system is part of the cached result. A later scan may prune a
        // different first row group, so it must not derive this origin from its current plan.
        int64_t base_granule = 0;
    };

    // Cache key for external tables (Hive ORC/Parquet)
    struct ExternalCacheKey {
        static constexpr uint8_t BASE_GRANULE_AWARE_VERSION = 1;

        ExternalCacheKey() = default;
        ExternalCacheKey(const std::string& path_, int64_t modification_time_, int64_t file_size_,
                         uint64_t digest_, int64_t start_offset_, int64_t size_,
                         uint8_t format_version_ = 0)
                : path(path_),
                  modification_time(modification_time_),
                  file_size(file_size_),
                  digest(digest_),
                  start_offset(start_offset_),
                  size(size_),
                  format_version(format_version_) {}
        std::string path;
        int64_t modification_time = 0;
        int64_t file_size = 0;
        uint64_t digest = 0;
        int64_t start_offset = 0;
        int64_t size = 0;
        uint8_t format_version = 0;

        [[nodiscard]] std::string encode() const {
            std::string key = path;
            char buf[41];
            memcpy(buf, &modification_time, 8);
            memcpy(buf + 8, &file_size, 8);
            memcpy(buf + 16, &digest, 8);
            memcpy(buf + 24, &start_offset, 8);
            memcpy(buf + 32, &size, 8);
            buf[40] = static_cast<char>(format_version);
            key.append(buf, 41);
            return key;
        }
    };

    // Create global instance of this class
    static ConditionCache* create_global_cache(size_t capacity, uint32_t num_shards = 16) {
        auto* res = new ConditionCache(capacity, num_shards);
        return res;
    }

    // Return global instance.
    // Client should call create_global_cache before.
    static ConditionCache* instance() { return ExecEnv::GetInstance()->get_condition_cache(); }

    ConditionCache() = delete;

    ConditionCache(size_t capacity, uint32_t num_shards)
            : LRUCachePolicy(CachePolicy::CacheType::CONDITION_CACHE, capacity, LRUCacheType::SIZE,
                             config::inverted_index_cache_stale_sweep_time_sec, num_shards,
                             /*element_count_capacity*/ 0, /*enable_prune*/ true,
                             /*is_lru_k*/ true) {}

    template <typename KeyType>
    bool lookup(const KeyType& key, ConditionCacheHandle* handle);

    template <typename KeyType>
    void insert(const KeyType& key, std::shared_ptr<std::vector<bool>> filter_result,
                int64_t base_granule = 0);
};

class ConditionCacheHandle {
public:
    ConditionCacheHandle() = default;

    ConditionCacheHandle(LRUCachePolicy* cache, Cache::Handle* handle)
            : _cache(cache), _handle(handle) {}

    ~ConditionCacheHandle() {
        if (_handle != nullptr) {
            _cache->release(_handle);
        }
    }

    ConditionCacheHandle(ConditionCacheHandle&& other) noexcept {
        // we can use std::exchange if we switch c++14 on
        std::swap(_cache, other._cache);
        std::swap(_handle, other._handle);
    }

    ConditionCacheHandle& operator=(ConditionCacheHandle&& other) noexcept {
        std::swap(_cache, other._cache);
        std::swap(_handle, other._handle);
        return *this;
    }

    LRUCachePolicy* cache() const { return _cache; }

    std::shared_ptr<std::vector<bool>> get_filter_result() const {
        if (!_cache) {
            return nullptr;
        }
        return ((ConditionCache::CacheValue*)_cache->value(_handle))->filter_result;
    }

    int64_t get_base_granule() const {
        DORIS_CHECK(_cache != nullptr);
        return ((ConditionCache::CacheValue*)_cache->value(_handle))->base_granule;
    }

private:
    LRUCachePolicy* _cache = nullptr;
    Cache::Handle* _handle = nullptr;

    // Don't allow copy and assign
    DISALLOW_COPY_AND_ASSIGN(ConditionCacheHandle);
};

} // namespace segment_v2
} // namespace doris