// 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 "storage/segment/column_reader_cache.h"

#include "storage/segment/column_meta_accessor.h"

#include "storage/segment/segment.h"

#include "storage/segment/variant/variant_column_reader.h"

bvar::Adder<int64_t> g_segment_column_reader_cache_count("segment_column_cache_count");

bvar::Adder<int64_t> g_segment_column_cache_hit_count("segment_column_cache_hit_count");

bvar::Adder<int64_t> g_segment_column_cache_miss_count("segment_column_cache_miss_count");

bvar::Adder<int64_t> g_segment_column_cache_evict_count("segment_column_cache_evict_count");

namespace doris::segment_v2 {

ColumnReaderCache::ColumnReaderCache(

        ColumnMetaAccessor* accessor, TabletSchemaSPtr tablet_schema,

        io::FileReaderSPtr file_reader, uint64_t num_rows,

        std::function<Status(std::shared_ptr<SegmentFooterPB>&, OlapReaderStatistics*)>

                get_footer_cb)

        : _accessor(accessor),

          _tablet_schema(std::move(tablet_schema)),

          _file_reader(std::move(file_reader)),

          _num_rows(num_rows),

          _get_footer_cb(std::move(get_footer_cb)) {}

ColumnReaderCache::~ColumnReaderCache() {

    g_segment_column_reader_cache_count << -_cache_map.size();

}

std::shared_ptr<ColumnReader> ColumnReaderCache::_lookup(const ColumnReaderCacheKey& key) {

    std::lock_guard<std::mutex> lock(_cache_mutex);

    auto it = _cache_map.find(key);

    if (it == _cache_map.end()) {

        g_segment_column_cache_miss_count << 1;

        return nullptr;

    }

    // Move the accessed node to the front of the linked list

    _lru_list.splice(_lru_list.begin(), _lru_list, it->second);

    DCHECK_EQ(it->second->key.first, key.first);

    g_segment_column_cache_hit_count << 1;

    return it->second->reader;

}

void ColumnReaderCache::_insert_locked_nocheck(const ColumnReaderCacheKey& key,

                                               const std::shared_ptr<ColumnReader>& reader) {

    // If capacity exceeded, remove least recently used (tail)

    if (_cache_map.size() >= config::max_segment_partial_column_cache_size) {

        g_segment_column_reader_cache_count << -1;

        g_segment_column_cache_evict_count << 1;

        auto last_it = _lru_list.end();

        --last_it;

        _cache_map.erase(last_it->key);

        _lru_list.pop_back();

    }

    g_segment_column_reader_cache_count << 1;

    _lru_list.push_front(CacheNode {

            .key = key, .reader = reader, .last_access = std::chrono::steady_clock::now()});

    _cache_map[key] = _lru_list.begin();

}

void ColumnReaderCache::_insert_direct(const ColumnReaderCacheKey& key,

                                       const std::shared_ptr<ColumnReader>& column_reader) {

    std::lock_guard<std::mutex> lock(_cache_mutex);

    _insert_locked_nocheck(key, column_reader);

}

std::map<int32_t, std::shared_ptr<ColumnReader>> ColumnReaderCache::get_available_readers(

        bool include_subcolumns) {

    std::lock_guard<std::mutex> lock(_cache_mutex);

    std::map<int32_t, std::shared_ptr<ColumnReader>> readers;

    for (const auto& node : _lru_list) {

        if (include_subcolumns || node.key.second.empty()) {

            readers.insert({node.key.first, node.reader});

        }

    }

    return readers;

}

Status ColumnReaderCache::get_column_reader(int32_t col_uid,

                                            std::shared_ptr<ColumnReader>* column_reader,

                                            OlapReaderStatistics* stats) {

    // Attempt to find in cache

    if (auto cached = _lookup({col_uid, {}})) {

        *column_reader = cached;

        return Status::OK();

    }

    // Load footer once under cache mutex (not thread-safe otherwise)

    std::shared_ptr<SegmentFooterPB> footer_pb_shared;

    {

        std::lock_guard<std::mutex> lock(_cache_mutex);

        RETURN_IF_ERROR(_get_footer_cb(footer_pb_shared, stats));

    }

    // Lookup column meta by uid via ColumnMetaAccessor. If not initialized or not found, return NOT_FOUND.

    ColumnMetaPB meta;

    Status st_meta = _accessor->get_column_meta_by_uid(*footer_pb_shared, col_uid, &meta);

    if (st_meta.is<ErrorCode::NOT_FOUND>()) {

        *column_reader = nullptr;

        return st_meta;

    }

    RETURN_IF_ERROR(st_meta);

    ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(),

                              .be_exec_version = _be_exec_version,

                              .tablet_schema = _tablet_schema};

    std::shared_ptr<ColumnReader> reader;

    if ((FieldType)meta.type() == FieldType::OLAP_FIELD_TYPE_VARIANT) {

        // Variant root columns require VariantColumnReader, which encapsulates

        // subcolumn layout, sparse columns and external meta.

        std::unique_ptr<VariantColumnReader> variant_reader(new VariantColumnReader());

        RETURN_IF_ERROR(variant_reader->init(opts, _accessor, footer_pb_shared, col_uid, _num_rows,

                                             _file_reader));

        reader.reset(variant_reader.release());

        VLOG_DEBUG << "insert cache (variant): uid=" << col_uid << " col_id=" << meta.column_id();

    } else {

        // For non-variant columns, we can create reader directly from ColumnMetaPB.

        VLOG_DEBUG << "insert cache: uid=" << col_uid << " col_id=" << meta.column_id();

        RETURN_IF_ERROR(ColumnReader::create(opts, meta, _num_rows, _file_reader, &reader));

    }

    _insert_direct({col_uid, {}}, reader);

    *column_reader = std::move(reader);

    return Status::OK();

}

Status ColumnReaderCache::get_path_column_reader(int32_t col_uid, PathInData relative_path,

                                                 std::shared_ptr<ColumnReader>* column_reader,

                                                 OlapReaderStatistics* stats,

                                                 const SubcolumnColumnMetaInfo::Node* node_hint) {

    // Attempt to find in cache first

    if (auto cached = _lookup({col_uid, relative_path})) {

        *column_reader = cached;

        return Status::OK();

    }

    if (!_accessor->has_column_uid(col_uid)) {

        *column_reader = nullptr;

        return Status::Error<ErrorCode::NOT_FOUND, false>("column not found in segment, col_uid={}",

                                                          col_uid);

    }

    // Load footer once under cache mutex (not thread-safe otherwise)

    std::shared_ptr<SegmentFooterPB> footer_pb_shared;

    {

        std::lock_guard<std::mutex> lock(_cache_mutex);

        RETURN_IF_ERROR(_get_footer_cb(footer_pb_shared, stats));

    }

    // Ensure variant root reader is available in cache.

    ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(),

                              .be_exec_version = _be_exec_version,

                              .tablet_schema = _tablet_schema};

    std::shared_ptr<ColumnReader> variant_column_reader;

    RETURN_IF_ERROR(get_column_reader(col_uid, &variant_column_reader, stats));

    if (relative_path.empty()) {

        *column_reader = std::move(variant_column_reader);

        return Status::OK();

    }

    // Delegate path-level reader creation to VariantColumnReader, which hides

    // inline vs external meta details.

    std::shared_ptr<ColumnReader> path_reader;

    auto* vreader = static_cast<VariantColumnReader*>(variant_column_reader.get());

    Status st = vreader->create_path_reader(relative_path, opts, _accessor, *footer_pb_shared,

                                            _file_reader, _num_rows, &path_reader);

    if (st.is<ErrorCode::NOT_FOUND>()) {

        *column_reader = nullptr;

        return st;

    }

    RETURN_IF_ERROR(st);

    // Cache and return

    _insert_direct({col_uid, relative_path}, path_reader);

    *column_reader = std::move(path_reader);

    return Status::OK();

}

} // namespace doris::segment_v2