// Licensed to the LF AI & Data foundation 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 <map>
#include <memory>
#include <string>
#include <roaring/roaring.hh>

#include "common/RegexQuery.h"
#include "index/ScalarIndex.h"
#include "pb/common.pb.h"
#include "storage/FileManager.h"
#include "storage/MemFileManagerImpl.h"

namespace milvus {
namespace index {

struct BitmapInfo {
    size_t offset_;
    size_t size_;
};

enum class BitmapIndexBuildMode {
    ROARING,
    BITSET,
};

/*
* @brief Implementation of Bitmap Index 
* @details This index only for scalar Integral type.
*/
template <typename T>
class BitmapIndex : public ScalarIndex<T> {
 public:
    explicit BitmapIndex(
        const storage::FileManagerContext& file_manager_context =
            storage::FileManagerContext(),
        bool is_nested_index = false);

    ~BitmapIndex() {
        if (is_mmap_) {
            UnmapIndexData();
        }
    }

    BinarySet
    Serialize(const Config& config) override;

    void
    Load(const BinarySet& index_binary, const Config& config = {}) override;

    void
    Load(milvus::tracer::TraceContext ctx, const Config& config = {}) override;

    int64_t
    Count() override {
        return total_num_rows_;
    }

    ScalarIndexType
    GetIndexType() const override {
        return ScalarIndexType::BITMAP;
    }

    // Reverse_Lookup is O(1) only when the offset cache is built; otherwise it
    // linearly scans all distinct postings (O(cardinality)) per row.
    bool
    SupportFastReverseLookup() const override {
        return use_offset_cache_;
    }

    bool
    IsNestedIndex() const override {
        return is_nested_index_;
    }

    void
    Build(size_t n, const T* values, const bool* valid_data = nullptr) override;

    void
    Build(const Config& config = {}) override;

    void
    BuildWithFieldData(const std::vector<FieldDataPtr>& datas) override;

    const TargetBitmap
    In(size_t n, const T* values) override;

    const TargetBitmap
    NotIn(size_t n, const T* values) override;

    const TargetBitmap
    IsNull() override;

    TargetBitmap
    IsNotNull() override;

    const TargetBitmap
    Range(const T& value, OpType op) override;

    const TargetBitmap
    Range(const T& lower_bound_value,
          bool lb_inclusive,
          const T& upper_bound_value,
          bool ub_inclusive) override;

    std::optional<T>
    Reverse_Lookup(size_t offset) const override;

    int64_t
    Size() override {
        return Count();
    }

    void
    ComputeByteSize() override {
        ScalarIndex<T>::ComputeByteSize();
        int64_t total = this->cached_byte_size_;

        // valid_bitset_
        total += valid_bitset_.size_in_bytes();

        if (is_mmap_) {
            // mmap mode
            total += mmap_size_;
            // bitmap_info_map_ overhead (keys and roaring metadata in memory)
            size_t num_entries = bitmap_info_map_.size();
            if constexpr (std::is_same_v<T, std::string>) {
                for (const auto& [key, bitmap] : bitmap_info_map_) {
                    total += key.capacity();
                }
            } else {
                total += num_entries * sizeof(T);
            }
            // roaring metadata + map node overhead per entry
            total += num_entries * (sizeof(roaring::Roaring) + 40);
        } else if (build_mode_ == BitmapIndexBuildMode::ROARING) {
            // data_: map<T, roaring::Roaring>
            for (const auto& [key, bitmap] : data_) {
                if constexpr (std::is_same_v<T, std::string>) {
                    total += key.capacity();
                } else {
                    total += sizeof(T);
                }
                total += bitmap.getSizeInBytes();
                // std::map red-black tree node overhead (~40 bytes per node)
                total += 40;
            }
        } else {
            // bitsets_: map<T, TargetBitmap>
            size_t num_entries = bitsets_.size();
            if (num_entries > 0) {
                size_t bitset_bytes = bitsets_.begin()->second.size_in_bytes();
                total += num_entries * bitset_bytes;
                if constexpr (std::is_same_v<T, std::string>) {
                    for (const auto& [key, bitset] : bitsets_) {
                        total += key.capacity();
                    }
                } else {
                    total += num_entries * sizeof(T);
                }
                // std::map red-black tree node overhead (~40 bytes per node)
                total += num_entries * 40;
            }
        }

        // offset cache
        total += data_offsets_cache_.capacity() *
                 sizeof(typename decltype(data_offsets_cache_)::value_type);
        total += bitsets_offsets_cache_.capacity() *
                 sizeof(typename decltype(bitsets_offsets_cache_)::value_type);
        total += mmap_offsets_cache_.capacity() *
                 sizeof(typename decltype(mmap_offsets_cache_)::value_type);

        this->cached_byte_size_ = total;
    }

    IndexStatsPtr
    Upload(const Config& config = {}) override;

    const bool
    HasRawData() const override {
        if (schema_.data_type() == proto::schema::DataType::Array) {
            return false;
        }
        return true;
    }

    void
    LoadWithoutAssemble(const BinarySet& binary_set,
                        const Config& config) override;

    const TargetBitmap
    Query(const DatasetPtr& dataset) override;

    void
    WriteEntries(storage::IndexEntryWriter* writer) override;

    void
    LoadEntries(storage::IndexEntryReader& reader,
                const Config& config) override;

    bool
    SupportPatternMatch() const override {
        return std::is_same_v<T, std::string>;
    }

    const TargetBitmap
    PatternMatch(const std::string& pattern, proto::plan::OpType op) override {
        switch (op) {
            case proto::plan::OpType::PrefixMatch:
            case proto::plan::OpType::PostfixMatch:
            case proto::plan::OpType::InnerMatch: {
                auto dataset = std::make_unique<Dataset>();
                dataset->Set(milvus::index::OPERATOR_TYPE, op);
                dataset->Set(milvus::index::MATCH_VALUE, pattern);
                return Query(std::move(dataset));
            }
            case proto::plan::OpType::Match: {
                return PatternQuery(pattern);
            }
            case proto::plan::OpType::RegexMatch: {
                if constexpr (!std::is_same_v<T, std::string>) {
                    ThrowInfo(ErrorCode::OpTypeInvalid,
                              "RegexMatch only supported for string type");
                    return TargetBitmap{};
                } else {
                    AssertInfo(is_built_, "index has not been built");
                    PartialRegexMatcher matcher(pattern);
                    TargetBitmap res(total_num_rows_, false);
                    if (is_mmap_) {
                        for (const auto& [key, bitmap] : bitmap_info_map_) {
                            if (matcher(key)) {
                                for (const auto& v : bitmap) {
                                    res.set(v);
                                }
                            }
                        }
                    } else if (build_mode_ == BitmapIndexBuildMode::ROARING) {
                        for (const auto& [key, bitmap] : data_) {
                            if (matcher(key)) {
                                for (const auto& v : bitmap) {
                                    res.set(v);
                                }
                            }
                        }
                    } else {
                        for (const auto& [key, bitset] : bitsets_) {
                            if (matcher(key)) {
                                res |= bitset;
                            }
                        }
                    }
                    return res;
                }
            }
            default:
                ThrowInfo(ErrorCode::OpTypeInvalid,
                          "not supported op type: {} for index PatternMatch",
                          op);
        }
    }

 protected:
    const TargetBitmap
    PatternQuery(const std::string& pattern) override {
        if constexpr (!std::is_same_v<T, std::string>) {
            ThrowInfo(ErrorCode::OpTypeInvalid,
                      "pattern query only supported for string type");
            return TargetBitmap{};
        }

        AssertInfo(is_built_, "index has not been built");

        LikePatternMatcher matcher(pattern);
        TargetBitmap res(total_num_rows_, false);
        if (is_mmap_) {
            for (const auto& [key, bitmap] : bitmap_info_map_) {
                if (matcher(key)) {
                    for (const auto& v : bitmap) {
                        res.set(v);
                    }
                }
            }
        } else if (build_mode_ == BitmapIndexBuildMode::ROARING) {
            for (const auto& [key, bitmap] : data_) {
                if (matcher(key)) {
                    for (const auto& v : bitmap) {
                        res.set(v);
                    }
                }
            }
        } else {
            for (const auto& [key, bitset] : bitsets_) {
                if (matcher(key)) {
                    res |= bitset;
                }
            }
        }
        return res;
    }

 public:
    int64_t
    Cardinality() {
        if (is_mmap_) {
            return bitmap_info_map_.size();
        }

        if (build_mode_ == BitmapIndexBuildMode::ROARING) {
            return data_.size();
        } else {
            return bitsets_.size();
        }
    }

 private:
    void
    BuildPrimitiveField(const std::vector<FieldDataPtr>& datas);

    void
    BuildArrayField(const std::vector<FieldDataPtr>& datas);

    void
    BuildArrayFieldNested(const std::vector<FieldDataPtr>& datas);

    size_t
    GetIndexDataSize();

    void
    SerializeIndexData(uint8_t* index_data_ptr);

    std::pair<std::shared_ptr<uint8_t[]>, size_t>
    SerializeValidBitsetData() const;

    std::pair<std::shared_ptr<uint8_t[]>, size_t>
    SerializeIndexMeta();

    std::pair<size_t, size_t>
    DeserializeIndexMeta(const uint8_t* data_ptr, size_t data_size);

    void
    DeserializeValidBitsetData(const uint8_t* data_ptr, size_t data_size);

    T
    ParseKey(const uint8_t** ptr);

    // Deserialize posting data.
    //
    // New bitmap index formats persist valid_bitset_, which is the
    // authoritative source of row validity. Legacy formats do not, so we may
    // rebuild validity from postings as a backward-compatibility fallback for
    // nullable fields. Non-nullable fields do not persist valid_bitset_ and
    // are treated as all-valid on load.
    //
    // Rebuilding validity from postings is lossy for ARRAY fields: empty
    // arrays have no element postings, so they cannot be distinguished from
    // null arrays during reconstruction.
    void
    DeserializeIndexData(const uint8_t* data_ptr,
                         size_t index_length,
                         bool rebuild_validity_from_postings);

    void
    BuildOffsetCache();

    T
    Reverse_Lookup_InCache(size_t idx) const;

    void
    ChooseIndexLoadMode(int64_t index_length);

    bool
    ShouldSkip(const T lower_value, const T upper_value, const OpType op);

    TargetBitmap
    ConvertRoaringToBitset(const roaring::Roaring& values);

    TargetBitmap
    RangeForRoaring(const T& value, OpType op);

    TargetBitmap
    RangeForBitset(const T& value, OpType op);

    TargetBitmap
    RangeForMmap(const T& value, OpType op);

    TargetBitmap
    RangeForRoaring(const T& lower_bound_value,
                    bool lb_inclusive,
                    const T& upper_bound_value,
                    bool ub_inclusive);

    TargetBitmap
    RangeForBitset(const T& lower_bound_value,
                   bool lb_inclusive,
                   const T& upper_bound_value,
                   bool ub_inclusive);

    TargetBitmap
    RangeForMmap(const T& lower_bound_value,
                 bool lb_inclusive,
                 const T& upper_bound_value,
                 bool ub_inclusive);

    // Build mmap-backed posting storage from serialized bitmap index data.
    //
    // New bitmap index formats persist valid_bitset_, which is the
    // authoritative source of row validity. Legacy formats do not, so we may
    // rebuild validity from postings as a backward-compatibility fallback for
    // nullable fields. Non-nullable fields do not persist valid_bitset_ and
    // are treated as all-valid on load.
    //
    // Rebuilding validity from postings is lossy for ARRAY fields: empty
    // arrays have no element postings, so they cannot be distinguished from
    // null arrays during reconstruction.
    void
    MMapIndexData(const std::string& filepath,
                  const uint8_t* data,
                  size_t data_size,
                  size_t index_length,
                  milvus::proto::common::LoadPriority priority,
                  bool rebuild_validity_from_postings);

    void
    UnmapIndexData();

 public:
    bool is_built_{false};
    BitmapIndexBuildMode build_mode_;
    std::map<T, roaring::Roaring> data_;
    std::map<T, TargetBitmap> bitsets_;
    bool is_mmap_{false};
    bool is_nested_index_{false};
    char* mmap_data_;
    int64_t mmap_size_;
    std::map<T, roaring::Roaring> bitmap_info_map_;
    size_t total_num_rows_{0};
    proto::schema::FieldSchema schema_;
    bool use_offset_cache_{false};
    std::vector<typename std::map<T, roaring::Roaring>::iterator>
        data_offsets_cache_;
    std::vector<typename std::map<T, TargetBitmap>::iterator>
        bitsets_offsets_cache_;
    std::vector<typename std::map<T, roaring::Roaring>::iterator>
        mmap_offsets_cache_;

    // generate valid_bitset to speed up NotIn and IsNull and IsNotNull operate
    TargetBitmap valid_bitset_;
};

}  // namespace index
}  // namespace milvus
