// 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/compaction/cumulative_compaction_policy.h"

#include <algorithm>

#include <iterator>

#include <list>

#include <ostream>

#include <string>

#include "common/config.h"

#include "common/logging.h"

#include "storage/compaction/cumulative_compaction_binlog_policy.h"

#include "storage/compaction/cumulative_compaction_time_series_policy.h"

#include "storage/olap_common.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_meta.h"

#include "util/debug_points.h"

#include "util/defer_op.h"

namespace doris {

SizeBasedCumulativeCompactionPolicy::SizeBasedCumulativeCompactionPolicy(

        int64_t promotion_size, double promotion_ratio, int64_t promotion_min_size,

        int64_t promotion_version_count, int64_t compaction_min_size)

        : _promotion_size(promotion_size),

          _promotion_ratio(promotion_ratio),

          _promotion_min_size(promotion_min_size),

          _promotion_version_count(promotion_version_count),

          _compaction_min_size(compaction_min_size) {}

void SizeBasedCumulativeCompactionPolicy::calculate_cumulative_point(

        Tablet* tablet, const RowsetMetaMapContainer& all_metas, int64_t current_cumulative_point,

        int64_t* ret_cumulative_point) {

    *ret_cumulative_point = Tablet::K_INVALID_CUMULATIVE_POINT;

    if (current_cumulative_point != Tablet::K_INVALID_CUMULATIVE_POINT) {

        // only calculate the point once.

        // after that, cumulative point will be updated along with compaction process.

        return;

    }

    // empty return

    if (all_metas.empty()) {

        return;

    }

    std::list<RowsetMetaSharedPtr> existing_rss;

    for (const auto& [_, rs] : all_metas) {

        existing_rss.emplace_back(rs);

    }

    // sort the existing rowsets by version in ascending order

    existing_rss.sort([](const RowsetMetaSharedPtr& a, const RowsetMetaSharedPtr& b) {

        // simple because 2 versions are certainly not overlapping

        return a->version().first < b->version().first;

    });

    // calculate promotion size

    auto base_rowset_meta = existing_rss.begin();

    if (tablet->tablet_state() == TABLET_RUNNING) {

        // check base rowset first version must be zero

        // for tablet which state is not TABLET_RUNNING, there may not have base version.

        CHECK((*base_rowset_meta)->start_version() == 0);

        int64_t promotion_size = 0;

        _calc_promotion_size(tablet, *base_rowset_meta, &promotion_size);

        int64_t prev_version = -1;

        for (const RowsetMetaSharedPtr& rs : existing_rss) {

            if (rs->version().first > prev_version + 1) {

                // There is a hole, do not continue

                break;

            }

            bool is_delete = rs->has_delete_predicate();

            // break the loop if segments in this rowset is overlapping.

            if (!is_delete && rs->is_segments_overlapping()) {

                *ret_cumulative_point = rs->version().first;

                break;

            }

            // check the rowset is whether less than promotion size

            if (!is_delete && rs->version().first != 0 && rs->total_disk_size() < promotion_size) {

                *ret_cumulative_point = rs->version().first;

                break;

            }

            // include one situation: When the segment is not deleted, and is singleton delta, and is NONOVERLAPPING, ret_cumulative_point increase

            prev_version = rs->version().second;

            *ret_cumulative_point = prev_version + 1;

        }

        VLOG_NOTICE

                << "cumulative compaction size_based policy, calculate cumulative point value = "

                << *ret_cumulative_point << ", calc promotion size value = " << promotion_size

                << " tablet = " << tablet->tablet_id();

    } else if (tablet->tablet_state() == TABLET_NOTREADY) {

        // tablet under alter process

        // we choose version next to the base version as cumulative point

        for (const RowsetMetaSharedPtr& rs : existing_rss) {

            if (rs->version().first > 0) {

                *ret_cumulative_point = rs->version().first;

                break;

            }

        }

    }

}

void SizeBasedCumulativeCompactionPolicy::_calc_promotion_size(Tablet* tablet,

                                                               RowsetMetaSharedPtr base_rowset_meta,

                                                               int64_t* promotion_size) {

    int64_t base_size = base_rowset_meta->total_disk_size();

    *promotion_size = int64_t(cast_set<double>(base_size) * _promotion_ratio);

    // promotion_size is between _promotion_size and _promotion_min_size

    if (*promotion_size >= _promotion_size) {

        *promotion_size = _promotion_size;

    } else if (*promotion_size <= _promotion_min_size) {

        *promotion_size = _promotion_min_size;

    }

    _refresh_tablet_promotion_size(tablet, *promotion_size);

}

void SizeBasedCumulativeCompactionPolicy::_refresh_tablet_promotion_size(Tablet* tablet,

                                                                         int64_t promotion_size) {

    tablet->set_cumulative_promotion_size(promotion_size);

}

void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(

        Tablet* tablet, const std::vector<RowsetSharedPtr>& input_rowsets,

        RowsetSharedPtr output_rowset, Version& last_delete_version) {

    if (tablet->tablet_state() != TABLET_RUNNING) {

        // if tablet under alter process, do not update cumulative point

        return;

    }

    // if rowsets have delete version, move to the last directly

    if (last_delete_version.first != -1) {

        tablet->set_cumulative_layer_point(output_rowset->end_version() + 1);

    } else {

        // if rowsets have no delete version, check output_rowset total disk size

        // satisfies promotion size.

        size_t total_size = output_rowset->rowset_meta()->total_disk_size();

        if (total_size >= tablet->cumulative_promotion_size()) {

            tablet->set_cumulative_layer_point(output_rowset->end_version() + 1);

        } else if (tablet->enable_unique_key_merge_on_write() &&

                   output_rowset->end_version() - output_rowset->start_version() >

                           _promotion_version_count) {

            // for MoW table, if there's too many versions, the delete bitmap will grow to

            // a very big size, which may cause the tablet meta too big and the `save_meta`

            // operation too slow.

            // if the rowset should not promotion according to it's disk size, we should also

            // consider it's version count here.

            tablet->set_cumulative_layer_point(output_rowset->end_version() + 1);

        }

    }

}

uint32_t SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(Tablet* tablet) {

    uint32_t score = 0;

    bool base_rowset_exist = false;

    const int64_t point = tablet->cumulative_layer_point();

    int64_t promotion_size = 0;

    std::vector<RowsetMetaSharedPtr> rowset_to_compact;

    int64_t total_size = 0;

    RowsetMetaSharedPtr first_meta;

    int64_t first_version = INT64_MAX;

    // NOTE: tablet._meta_lock is hold

    auto& rs_metas = tablet->tablet_meta()->all_rs_metas();

    // check the base rowset and collect the rowsets of cumulative part

    for (const auto& [_, rs_meta] : rs_metas) {

        if (rs_meta->start_version() < first_version) {

            first_version = rs_meta->start_version();

            first_meta = rs_meta;

        }

        // check base rowset

        if (rs_meta->start_version() == 0) {

            base_rowset_exist = true;

        }

        if (rs_meta->end_version() < point || !rs_meta->is_local()) {

            // all_rs_metas() is not sorted, so we use _continue_ other than _break_ here.

            continue;

        } else {

            // collect the rowsets of cumulative part

            total_size += rs_meta->total_disk_size();

            score += rs_meta->get_compaction_score();

            rowset_to_compact.push_back(rs_meta);

        }

    }

    if (first_meta == nullptr) {

        return 0;

    }

    // Use "first"(not base) version to calc promotion size

    // because some tablet do not have base version(under alter operation)

    _calc_promotion_size(tablet, first_meta, &promotion_size);

    // If base version does not exist, but its state is RUNNING.

    // It is abnormal, do not select it and set *score = 0

    if (!base_rowset_exist && tablet->tablet_state() == TABLET_RUNNING) {

        LOG(WARNING) << "tablet state is running but have no base version";

        return 0;

    }

    // if total_size is greater than promotion_size, return total score

    if (total_size >= promotion_size) {

        return score;

    }

    // sort the rowsets of cumulative part

    std::sort(rowset_to_compact.begin(), rowset_to_compact.end(), RowsetMeta::comparator);

    // calculate the rowsets to do cumulative compaction

    // eg: size of rowset_to_compact are:

    // 128, 16, 16, 16

    // we will choose [16,16,16] to compact.

    for (auto& rs_meta : rowset_to_compact) {

        int64_t current_level = _level_size(rs_meta->total_disk_size());

        int64_t remain_level = _level_size(total_size - rs_meta->total_disk_size());

        // if current level less then remain level, score contains current rowset

        // and process return; otherwise, score does not contains current rowset.

        if (current_level <= remain_level) {

            return score;

        }

        total_size -= rs_meta->total_disk_size();

        score -= rs_meta->get_compaction_score();

    }

    return score;

}

int SizeBasedCumulativeCompactionPolicy::pick_input_rowsets(

        Tablet* tablet, const std::vector<RowsetSharedPtr>& candidate_rowsets,

        const int64_t max_compaction_score, const int64_t min_compaction_score,

        std::vector<RowsetSharedPtr>* input_rowsets, Version* last_delete_version,

        size_t* compaction_score, bool allow_delete) {

    DBUG_EXECUTE_IF("SizeBasedCumulativeCompactionPolicy::pick_input_rowsets.set_input_rowsets", {

        auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);

        if (target_tablet_id == tablet->tablet_id()) {

            auto start_version = dp->param<int64_t>("start_version", -1);

            auto end_version = dp->param<int64_t>("end_version", -1);

            for (auto& rowset : candidate_rowsets) {

                if (rowset->start_version() >= start_version &&

                    rowset->end_version() <= end_version) {

                    input_rowsets->push_back(rowset);

                }

            }

        }

        return cast_set<uint32_t>(input_rowsets->size());

    })

    size_t promotion_size = tablet->cumulative_promotion_size();

    auto max_version = tablet->max_version().first;

    int transient_size = 0;

    *compaction_score = 0;

    int64_t total_size = 0;

    RowsetSharedPtr last_popped;

    // DEFER: trim input_rowsets from back if score > max_compaction_score

    // This ensures we don't return more rowsets than allowed by max_compaction_score,

    // while still collecting enough rowsets to pass min_compaction_score check after level_size removal.

    // Must be placed after variable initialization and before collection loop.

    DEFER({

        // Keep at least 1 rowset to avoid removing the only rowset (consistent with fallback branch)

        while (input_rowsets->size() > 1 &&

               *compaction_score > static_cast<size_t>(max_compaction_score)) {

            last_popped = std::move(input_rowsets->back());

            *compaction_score -= last_popped->rowset_meta()->get_compaction_score();

            total_size -= last_popped->rowset_meta()->total_disk_size();

            input_rowsets->pop_back();

        }

        // A single non-overlapping rowset cannot be compacted by itself. Restore the direct

        // successor and accept a one-off max-score overshoot to keep the input mergeable.

        if (input_rowsets->size() == 1 && last_popped != nullptr &&

            !input_rowsets->front()->rowset_meta()->is_segments_overlapping()) {

            *compaction_score += last_popped->rowset_meta()->get_compaction_score();

            total_size += last_popped->rowset_meta()->total_disk_size();

            input_rowsets->push_back(std::move(last_popped));

        }

    });

    for (auto& rowset : candidate_rowsets) {

        // check whether this rowset is delete version

        if (!allow_delete && rowset->rowset_meta()->has_delete_predicate()) {

            *last_delete_version = rowset->version();

            if (!input_rowsets->empty()) {

                // we meet a delete version, and there were other versions before.

                // we should compact those version before handling them over to base compaction

                break;

            } else {

                // we meet a delete version, and no other versions before, skip it and continue

                input_rowsets->clear();

                *compaction_score = 0;

                transient_size = 0;

                continue;

            }

        }

        if (tablet->tablet_state() == TABLET_NOTREADY) {

            // If tablet under alter, keep latest 10 version so that base tablet max version

            // not merged in new tablet, and then we can copy data from base tablet

            if (rowset->version().second > max_version - 10) {

                continue;

            }

        }

        // Removed: max_compaction_score check here

        // We now collect all candidate rowsets and trim from back at return time via DEFER

        *compaction_score += rowset->rowset_meta()->get_compaction_score();

        total_size += rowset->rowset_meta()->total_disk_size();

        transient_size += 1;

        input_rowsets->push_back(rowset);

    }

    DBUG_EXECUTE_IF("SizeBaseCumulativeCompactionPolicy.pick_input_rowsets.return_input_rowsets",

                    { return transient_size; })

    // if there is delete version, do compaction directly

    if (last_delete_version->first != -1) {

        if (input_rowsets->size() == 1) {

            auto rs_meta = input_rowsets->front()->rowset_meta();

            // if there is only one rowset and not overlapping,

            // we do not need to do cumulative compaction

            if (!rs_meta->is_segments_overlapping()) {

                input_rowsets->clear();

                *compaction_score = 0;

            }

        }

        return transient_size;

    }

    auto rs_begin = input_rowsets->begin();

    size_t new_compaction_score = *compaction_score;

    const bool can_handle_exhausted_input =

            *compaction_score >= static_cast<size_t>(max_compaction_score);

    while (rs_begin != input_rowsets->end()) {

        auto& rs_meta = (*rs_begin)->rowset_meta();

        int64_t current_level = _level_size(rs_meta->total_disk_size());

        int64_t remain_level = _level_size(total_size - rs_meta->total_disk_size());

        // if current level less then remain level, input rowsets contain current rowset

        // and process return; otherwise, input rowsets do not contain current rowset.

        if (current_level <= remain_level) {

            break;

        }

        auto next = std::next(rs_begin);

        // Keep the last suffix rowset for the singleton checks unless the exhausted-input

        // fallback below can select a useful input.

        if (next == input_rowsets->end() && !can_handle_exhausted_input) {

            break;

        }

        total_size -= rs_meta->total_disk_size();

        new_compaction_score -= rs_meta->get_compaction_score();

        rs_begin = next;

    }

    if (rs_begin == input_rowsets->end() && *compaction_score >= max_compaction_score) {

        // No suitable level size found in `input_rowsets` but score of `input_rowsets` exceed max compaction score,

        // which means `input_rowsets` will never change and this tablet will never execute cumulative compaction.

        // MUST execute compaction on these `input_rowsets` to reduce compaction score.

        RowsetSharedPtr rs_with_max_score;

        uint32_t max_score = 1;

        for (auto& rs : *input_rowsets) {

            if (rs->rowset_meta()->get_compaction_score() > max_score) {

                max_score = rs->rowset_meta()->get_compaction_score();

                rs_with_max_score = rs;

            }

        }

        if (rs_with_max_score) {

            input_rowsets->clear();

            input_rowsets->push_back(std::move(rs_with_max_score));

            *compaction_score = max_score;

            return transient_size;

        }

        // no rowset is OVERLAPPING, return all input rowsets (DEFER will trim to max_compaction_score)

        return transient_size;

    }

    input_rowsets->erase(input_rowsets->begin(), rs_begin);

    *compaction_score = new_compaction_score;

    VLOG_CRITICAL << "cumulative compaction size_based policy, compaction_score = "

                  << *compaction_score << ", total_size = " << total_size

                  << ", calc promotion size value = " << promotion_size

                  << ", tablet = " << tablet->tablet_id() << ", input_rowset size "

                  << input_rowsets->size();

    // empty return

    if (input_rowsets->empty()) {

        return transient_size;

    }

    // if we have a sufficient number of segments, we should process the compaction.

    // otherwise, we check number of segments and total_size whether can do compaction.

    if (total_size < _compaction_min_size && *compaction_score < min_compaction_score) {

        input_rowsets->clear();

        *compaction_score = 0;

    } else if (total_size >= _compaction_min_size && input_rowsets->size() == 1) {

        auto rs_meta = input_rowsets->front()->rowset_meta();

        // if there is only one rowset and not overlapping,

        // we do not need to do compaction

        if (!rs_meta->is_segments_overlapping()) {

            input_rowsets->clear();

            *compaction_score = 0;

        }

    }

    return transient_size;

}

int64_t SizeBasedCumulativeCompactionPolicy::_level_size(const int64_t size) {

    if (size < 1024) return 0;

    int64_t max_level = (int64_t)1

                        << (sizeof(_promotion_size) * 8 - 1 - __builtin_clzl(_promotion_size / 2));

    if (size >= max_level) return max_level;

    return (int64_t)1 << (sizeof(size) * 8 - 1 - __builtin_clzl(size));

}

std::shared_ptr<CumulativeCompactionPolicy>

CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(

        const std::string_view& compaction_policy) {

    if (compaction_policy == CUMULATIVE_TIME_SERIES_POLICY) {

        return std::make_shared<TimeSeriesCumulativeCompactionPolicy>();

    } else if (compaction_policy == CUMULATIVE_SIZE_BASED_POLICY) {

        return std::make_shared<SizeBasedCumulativeCompactionPolicy>();

    } else if (compaction_policy == CUMULATIVE_BINLOG_POLICY) {

        return std::make_shared<BinlogCumulativeCompactionPolicy>();

    }

    return std::make_shared<SizeBasedCumulativeCompactionPolicy>();

}

} // namespace doris