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

#ifndef DORIS_BE_SRC_OLAP_ROWSET_ROWSET_META_H

#define DORIS_BE_SRC_OLAP_ROWSET_ROWSET_META_H

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <atomic>

#include <chrono>

#include <cstdint>

#include <memory>

#include <string>

#include <vector>

#include "common/cast_set.h"

#include "common/config.h"

#include "common/status.h"

#include "io/fs/encrypted_fs_factory.h"

#include "io/fs/file_system.h"

#include "runtime/memory/lru_cache_policy.h"

#include "storage/metadata_adder.h"

#include "storage/olap_common.h"

#include "storage/rowset/rowset_fwd.h"

#include "storage/storage_policy.h"

#include "storage/tablet/tablet_fwd.h"

#include "util/once.h"

namespace doris {

class RowsetMeta : public MetadataAdder<RowsetMeta> {

public:

    RowsetMeta() = default;

    ~RowsetMeta();

    bool init(std::string_view pb_rowset_meta);

    bool init(const RowsetMeta* rowset_meta);

    bool init_from_pb(const RowsetMetaPB& rowset_meta_pb);

    bool init_from_json(const std::string& json_rowset_meta);

    bool serialize(std::string* value) { return _serialize_to_pb(value); }

    bool json_rowset_meta(std::string* json_rowset_meta);

    // If the rowset is a local rowset, return the global local file system.

    // Otherwise, return the remote file system corresponding to rowset's resource id.

    // Note that if the resource id cannot be found for the corresponding remote file system, nullptr will be returned.

    MOCK_FUNCTION io::FileSystemSPtr fs();

    io::FileSystemSPtr physical_fs();

    Result<const StorageResource*> remote_storage_resource();

    void set_remote_storage_resource(StorageResource resource);

    const std::string& resource_id() const { return _rowset_meta_pb.resource_id(); }

    void set_resource_id(const std::string& resource_id) {

        _rowset_meta_pb.set_resource_id(resource_id);

    }

    bool is_local() const { return !_rowset_meta_pb.has_resource_id(); }

    bool has_variant_type_in_schema() const;

    RowsetId rowset_id() const { return _rowset_id; }

    void set_rowset_id(const RowsetId& rowset_id) {

        // rowset id is a required field, just set it to 0

        _rowset_meta_pb.set_rowset_id(0);

        _rowset_id = rowset_id;

        _rowset_meta_pb.set_rowset_id_v2(rowset_id.to_string());

    }

    int64_t tablet_id() const { return _rowset_meta_pb.tablet_id(); }

    void set_tablet_id(int64_t tablet_id) { _rowset_meta_pb.set_tablet_id(tablet_id); }

    int64_t index_id() const { return _rowset_meta_pb.index_id(); }

    void set_index_id(int64_t index_id) { _rowset_meta_pb.set_index_id(index_id); }

    TabletUid tablet_uid() const { return _rowset_meta_pb.tablet_uid(); }

    void set_tablet_uid(TabletUid tablet_uid) {

        *(_rowset_meta_pb.mutable_tablet_uid()) = tablet_uid.to_proto();

    }

    int64_t txn_id() const { return _rowset_meta_pb.txn_id(); }

    void set_txn_id(int64_t txn_id) { _rowset_meta_pb.set_txn_id(txn_id); }

    int32_t tablet_schema_hash() const { return _rowset_meta_pb.tablet_schema_hash(); }

    void set_tablet_schema_hash(int32_t tablet_schema_hash) {

        _rowset_meta_pb.set_tablet_schema_hash(tablet_schema_hash);

    }

    void mark_row_binlog() { _rowset_meta_pb.set_is_row_binlog(true); }

    bool is_row_binlog() const {

        return _rowset_meta_pb.has_is_row_binlog() && _rowset_meta_pb.is_row_binlog();

    }

    RowsetTypePB rowset_type() const { return _rowset_meta_pb.rowset_type(); }

    void set_rowset_type(RowsetTypePB rowset_type) { _rowset_meta_pb.set_rowset_type(rowset_type); }

    RowsetStatePB rowset_state() const { return _rowset_meta_pb.rowset_state(); }

    void set_rowset_state(RowsetStatePB rowset_state) {

        _rowset_meta_pb.set_rowset_state(rowset_state);

    }

    Version version() const {

        return {_rowset_meta_pb.start_version(), _rowset_meta_pb.end_version()};

    }

    void set_version(Version version) {

        _rowset_meta_pb.set_start_version(version.first);

        _rowset_meta_pb.set_end_version(version.second);

    }

    bool has_version() const {

        return _rowset_meta_pb.has_start_version() && _rowset_meta_pb.has_end_version();

    }

    int64_t start_version() const { return _rowset_meta_pb.start_version(); }

    int64_t end_version() const { return _rowset_meta_pb.end_version(); }

    int64_t num_rows() const { return _rowset_meta_pb.num_rows(); }

    void set_num_rows(int64_t num_rows) { _rowset_meta_pb.set_num_rows(num_rows); }

    void set_num_segment_rows(const std::vector<uint32_t>& num_segment_rows) {

        _rowset_meta_pb.mutable_num_segment_rows()->Assign(num_segment_rows.cbegin(),

                                                           num_segment_rows.cend());

    }

    void get_num_segment_rows(std::vector<uint32_t>* num_segment_rows) const {

        num_segment_rows->assign(_rowset_meta_pb.num_segment_rows().cbegin(),

                                 _rowset_meta_pb.num_segment_rows().cend());

    }

    auto& get_num_segment_rows() const { return _rowset_meta_pb.num_segment_rows(); }

    int64_t total_disk_size() const { return _rowset_meta_pb.total_disk_size(); }

    void set_total_disk_size(int64_t total_disk_size) {

        _rowset_meta_pb.set_total_disk_size(total_disk_size);

    }

    int64_t data_disk_size() const { return _rowset_meta_pb.data_disk_size(); }

    void set_data_disk_size(int64_t data_disk_size) {

        _rowset_meta_pb.set_data_disk_size(data_disk_size);

    }

    int64_t index_disk_size() const { return _rowset_meta_pb.index_disk_size(); }

    void set_index_disk_size(int64_t index_disk_size) {

        _rowset_meta_pb.set_index_disk_size(index_disk_size);

    }

    void zone_maps(std::vector<ZoneMap>* zone_maps) {

        for (const ZoneMap& zone_map : _rowset_meta_pb.zone_maps()) {

            zone_maps->push_back(zone_map);

        }

    }

    void set_zone_maps(const std::vector<ZoneMap>& zone_maps) {

        for (const ZoneMap& zone_map : zone_maps) {

            ZoneMap* new_zone_map = _rowset_meta_pb.add_zone_maps();

            *new_zone_map = zone_map;

        }

    }

    void add_zone_map(const ZoneMap& zone_map) {

        ZoneMap* new_zone_map = _rowset_meta_pb.add_zone_maps();

        *new_zone_map = zone_map;

    }

    bool has_delete_predicate() const { return _rowset_meta_pb.has_delete_predicate(); }

    const DeletePredicatePB& delete_predicate() const { return _rowset_meta_pb.delete_predicate(); }

    DeletePredicatePB* mutable_delete_predicate() {

        return _rowset_meta_pb.mutable_delete_predicate();

    }

    void set_delete_predicate(DeletePredicatePB delete_predicate) {

        DeletePredicatePB* new_delete_condition = _rowset_meta_pb.mutable_delete_predicate();

        *new_delete_condition = std::move(delete_predicate);

    }

    bool empty() const { return _rowset_meta_pb.empty(); }

    void set_empty(bool empty) { _rowset_meta_pb.set_empty(empty); }

    PUniqueId load_id() const { return _rowset_meta_pb.load_id(); }

    void set_load_id(PUniqueId load_id) {

        PUniqueId* new_load_id = _rowset_meta_pb.mutable_load_id();

        new_load_id->set_hi(load_id.hi());

        new_load_id->set_lo(load_id.lo());

    }

    void set_job_id(const std::string& job_id) { _rowset_meta_pb.set_job_id(job_id); }

    const std::string& job_id() const { return _rowset_meta_pb.job_id(); }

    bool delete_flag() const { return _rowset_meta_pb.delete_flag(); }

    int64_t creation_time() const { return _rowset_meta_pb.creation_time(); }

    void set_creation_time(int64_t creation_time) {

        return _rowset_meta_pb.set_creation_time(creation_time);

    }

    int64_t stale_at() const {

        int64_t stale_time = _stale_at_s.load();

        return stale_time > 0 ? stale_time : _rowset_meta_pb.creation_time();

    }

    bool has_stale_at() const { return _stale_at_s.load() > 0; }

    void set_stale_at(int64_t stale_at) { _stale_at_s.store(stale_at); }

    int64_t partition_id() const { return _rowset_meta_pb.partition_id(); }

    void set_partition_id(int64_t partition_id) {

        return _rowset_meta_pb.set_partition_id(partition_id);

    }

    int64_t num_segments() const { return _rowset_meta_pb.num_segments(); }

    void set_num_segments(int64_t num_segments) { _rowset_meta_pb.set_num_segments(num_segments); }

    // Convert to RowsetMetaPB, skip_schema is only used by cloud to separate schema from rowset meta.

    void to_rowset_pb(RowsetMetaPB* rs_meta_pb, bool skip_schema = false) const;

    // Convert to RowsetMetaPB, skip_schema is only used by cloud to separate schema from rowset meta.

    RowsetMetaPB get_rowset_pb(bool skip_schema = false) const;

    inline DeletePredicatePB* mutable_delete_pred_pb() {

        return _rowset_meta_pb.mutable_delete_predicate();

    }

    bool is_singleton_delta() const {

        return has_version() && _rowset_meta_pb.start_version() == _rowset_meta_pb.end_version();

    }

    // Some time, we may check if this rowset is in rowset meta manager's meta by using RowsetMetaManager::check_rowset_meta.

    // But, this check behavior may cost a lot of time when it is frequent.

    // If we explicitly remove this rowset from rowset meta manager's meta, we can set _is_removed_from_rowset_meta to true,

    // And next time when we want to check if this rowset is in rowset mata manager's meta, we can

    // check is_remove_from_rowset_meta() first.

    void set_remove_from_rowset_meta() { _is_removed_from_rowset_meta = true; }

    bool is_remove_from_rowset_meta() const { return _is_removed_from_rowset_meta; }

    SegmentsOverlapPB segments_overlap() const { return _rowset_meta_pb.segments_overlap_pb(); }

    void set_segments_overlap(SegmentsOverlapPB segments_overlap) {

        _rowset_meta_pb.set_segments_overlap_pb(segments_overlap);

    }

    static bool comparator(const RowsetMetaSharedPtr& left, const RowsetMetaSharedPtr& right) {

        return left->end_version() < right->end_version();

    }

    // return true if segments in this rowset has overlapping data.

    // this is not same as `segments_overlap()` method.

    // `segments_overlap()` only return the value of "segments_overlap" field in rowset meta,

    // but "segments_overlap" may be UNKNOWN.

    //

    // Returns true if all of the following conditions are met

    // 1. the rowset contains more than one segment

    // 2. the rowset's start version == end version (non-singleton rowset was generated by compaction process

    //    which always produces non-overlapped segments)

    // 3. segments_overlap() flag is not NONOVERLAPPING (OVERLAP_UNKNOWN and OVERLAPPING are OK)

    bool is_segments_overlapping() const {

        return num_segments() > 1 && is_singleton_delta() && segments_overlap() != NONOVERLAPPING;

    }

    bool produced_by_compaction() const {

        return has_version() &&

               (start_version() < end_version() ||

                (start_version() == end_version() && segments_overlap() == NONOVERLAPPING));

    }

    // get the compaction score of this rowset.

    // if segments are overlapping, the score equals to the number of segments,

    // otherwise, score is 1.

    uint32_t get_compaction_score() const {

        uint32_t score = 0;

        if (!is_segments_overlapping()) {

            score = 1;

        } else {

            auto num_seg = num_segments();

            DCHECK_GT(num_seg, 0);

            score = cast_set<uint32_t>(num_seg);

            CHECK(score > 0);

        }

        return score;

    }

    uint32_t get_merge_way_num() const {

        uint32_t way_num = 0;

        if (!is_segments_overlapping()) {

            if (num_segments() == 0) {

                way_num = 0;

            } else {

                way_num = 1;

            }

        } else {

            auto num_seg = num_segments();

            DCHECK_GT(num_seg, 0);

            way_num = cast_set<uint32_t>(num_seg);

            CHECK(way_num > 0);

        }

        return way_num;

    }

    void get_segments_key_bounds(std::vector<KeyBoundsPB>* segments_key_bounds) const {

        for (const KeyBoundsPB& key_range : _rowset_meta_pb.segments_key_bounds()) {

            segments_key_bounds->push_back(key_range);

        }

    }

    auto& get_segments_key_bounds() const { return _rowset_meta_pb.segments_key_bounds(); }

    bool is_segments_key_bounds_truncated() const {

        return _rowset_meta_pb.has_segments_key_bounds_truncated() &&

               _rowset_meta_pb.segments_key_bounds_truncated();

    }

    void set_segments_key_bounds_truncated(bool truncated) {

        _rowset_meta_pb.set_segments_key_bounds_truncated(truncated);

    }

    // When true, `segments_key_bounds` holds a single aggregated

    // [rowset_min, rowset_max] entry instead of per-segment bounds.

    bool is_segments_key_bounds_aggregated() const {

        return _rowset_meta_pb.has_segments_key_bounds_aggregated() &&

               _rowset_meta_pb.segments_key_bounds_aggregated();

    }

    void set_segments_key_bounds_aggregated(bool aggregated) {

        _rowset_meta_pb.set_segments_key_bounds_aggregated(aggregated);

    }

    bool get_first_segment_key_bound(KeyBoundsPB* key_bounds) {

        // for compatibility, old version has not segment key bounds

        if (_rowset_meta_pb.segments_key_bounds_size() == 0) {

            return false;

        }

        *key_bounds = *_rowset_meta_pb.segments_key_bounds().begin();

        return true;

    }

    bool get_last_segment_key_bound(KeyBoundsPB* key_bounds) {

        if (_rowset_meta_pb.segments_key_bounds_size() == 0) {

            return false;

        }

        *key_bounds = *_rowset_meta_pb.segments_key_bounds().rbegin();

        return true;

    }

    // If `aggregate_into_single` is true, collapse per-segment bounds into a single

    // [rowset_min, rowset_max] entry and mark this rowset as aggregated.

    void set_segments_key_bounds(const std::vector<KeyBoundsPB>& segments_key_bounds,

                                 bool aggregate_into_single = false);

    void add_segment_key_bounds(KeyBoundsPB segments_key_bounds) {

        *_rowset_meta_pb.add_segments_key_bounds() = std::move(segments_key_bounds);

        set_segments_overlap(OVERLAPPING);

    }

    void set_newest_write_timestamp(int64_t timestamp) {

        _rowset_meta_pb.set_newest_write_timestamp(timestamp);

    }

    int64_t newest_write_timestamp() const { return _rowset_meta_pb.newest_write_timestamp(); }

    // for cloud only

    bool has_visible_ts_ms() const { return _rowset_meta_pb.has_visible_ts_ms(); }

    int64_t visible_ts_ms() const { return _rowset_meta_pb.visible_ts_ms(); }

    std::chrono::time_point<std::chrono::system_clock> visible_timestamp() const {

        using namespace std::chrono;

        if (has_visible_ts_ms()) {

            return time_point<system_clock>(milliseconds(visible_ts_ms()));

        }

        return system_clock::from_time_t(newest_write_timestamp());

    }

    void set_visible_ts_ms(int64_t visible_ts_ms) {

        _rowset_meta_pb.set_visible_ts_ms(visible_ts_ms);

    }

    void set_tablet_schema(const TabletSchemaSPtr& tablet_schema);

    void set_tablet_schema(const TabletSchemaPB& tablet_schema);

    const TabletSchemaSPtr& tablet_schema() const { return _schema; }

    void set_txn_expiration(int64_t expiration) { _rowset_meta_pb.set_txn_expiration(expiration); }

    void set_compaction_level(int64_t compaction_level) {

        _rowset_meta_pb.set_compaction_level(compaction_level);

    }

    int64_t compaction_level() { return _rowset_meta_pb.compaction_level(); }

    // `seg_file_size` MUST ordered by segment id

    void add_segments_file_size(const std::vector<size_t>& seg_file_size);

    // Return -1 if segment file size is unknown

    int64_t segment_file_size(int seg_id) const;

    const auto& segments_file_size() const { return _rowset_meta_pb.segments_file_size(); }

    // Used for partial update, when publish, partial update may add a new rowset and we should update rowset meta

    void merge_rowset_meta(const RowsetMeta& other);

    InvertedIndexFileInfo inverted_index_file_info(int seg_id);

    const auto& inverted_index_file_info() const {

        return _rowset_meta_pb.inverted_index_file_info();

    }

    void add_inverted_index_files_info(

            const std::vector<const InvertedIndexFileInfo*>& idx_file_info);

    int64_t get_metadata_size() const override;

    // Because the member field '_handle' is a raw pointer, use member func 'init' to replace copy ctor

    RowsetMeta(const RowsetMeta&) = delete;

    RowsetMeta operator=(const RowsetMeta&) = delete;

    void add_packed_slice_location(const std::string& segment_path,

                                   const std::string& packed_file_path, int64_t offset,

                                   int64_t size, int64_t packed_file_size) {

        auto* index_map = _rowset_meta_pb.mutable_packed_slice_locations();

        auto& index_pb = (*index_map)[segment_path];

        index_pb.set_packed_file_path(packed_file_path);

        index_pb.set_offset(offset);

        index_pb.set_size(size);

        index_pb.set_packed_file_size(packed_file_size);

    }

    int32_t schema_version() const { return _rowset_meta_pb.schema_version(); }

    std::string debug_string() const { return _rowset_meta_pb.ShortDebugString(); }

    // Pre-set the encryption algorithm to avoid re-entrant get_tablet calls

    // that can cause SingleFlight deadlock during tablet loading.

    void set_encryption_algorithm(EncryptionAlgorithmPB algorithm) {

        _determine_encryption_once.call(

                [algorithm]() -> Result<EncryptionAlgorithmPB> { return algorithm; });

    }

    int64_t commit_tso() const { return _rowset_meta_pb.commit_tso(); }

    void set_commit_tso(int64_t commit_tso) { _rowset_meta_pb.set_commit_tso(commit_tso); }

    void set_cloud_fields_after_visible(int64_t visible_version, int64_t version_update_time_ms) {

        // Update rowset meta with correct version and visible_ts

        // !!ATTENTION!!: this code should be updated if there are more fields

        // in rowset meta which will be modified in meta-service when commit_txn in the future

        set_version({visible_version, visible_version});

        if (version_update_time_ms > 0) {

            set_visible_ts_ms(version_update_time_ms);

        }

    }

private:

    bool _deserialize_from_pb(std::string_view value);

    bool _serialize_to_pb(std::string* value);

    void _init();

    friend bool operator==(const RowsetMeta& a, const RowsetMeta& b);

    friend bool operator!=(const RowsetMeta& a, const RowsetMeta& b) { return !(a == b); }

private:

    RowsetMetaPB _rowset_meta_pb;

    TabletSchemaSPtr _schema;

    Cache::Handle* _handle = nullptr;

    RowsetId _rowset_id;

    StorageResource _storage_resource;

    bool _is_removed_from_rowset_meta = false;

    DorisCallOnce<Result<EncryptionAlgorithmPB>> _determine_encryption_once;

    std::atomic<int64_t> _stale_at_s {0};

};

using RowsetMetaMapContainer = std::unordered_map<Version, RowsetMetaSharedPtr, HashOfVersion>;

} // namespace doris

#endif // DORIS_BE_SRC_OLAP_ROWSET_ROWSET_META_H