// 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 "cloud/cloud_cumulative_compaction_policy.h"

#include <algorithm>

#include <iterator>

#include <list>

#include <ostream>

#include <string>

#include "cloud/config.h"

#include "common/config.h"

#include "common/logging.h"

#include "cpp/sync_point.h"

#include "storage/compaction/cumulative_compaction_time_series_policy.h"

#include "storage/olap_common.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_meta.h"

#include "util/defer_op.h"

namespace doris {

int64_t CloudCumulativeCompactionPolicy::calculate_cumulative_point(

        CloudTablet* tablet, const std::vector<RowsetSharedPtr>& rowsets,

        const RowsetSharedPtr& output_rowset, Version& last_delete_version,

        int64_t input_cumulative_point) {

    DORIS_CHECK(std::find(rowsets.begin(), rowsets.end(), output_rowset) != rowsets.end());

    int64_t cumulative_point = input_cumulative_point;

    Version no_delete_version {-1, -1};

    for (const auto& rowset : rowsets) {

        DORIS_CHECK_EQ(rowset->start_version(), cumulative_point);

        if (rowset->rowset_meta()->has_delete_predicate()) {

            cumulative_point = rowset->end_version() + 1;

            continue;

        }

        if (rowset->rowset_meta()->is_segments_overlapping()) {

            return cumulative_point;

        }

        Version& delete_version = rowset == output_rowset ? last_delete_version : no_delete_version;

        int64_t candidate_cumulative_point =

                new_cumulative_point(tablet, rowset, delete_version, cumulative_point);

        if (candidate_cumulative_point != rowset->end_version() + 1) {

            return cumulative_point;

        }

        cumulative_point = candidate_cumulative_point;

    }

    return cumulative_point;

}

CloudSizeBasedCumulativeCompactionPolicy::CloudSizeBasedCumulativeCompactionPolicy(

        int64_t promotion_size, double promotion_ratio, int64_t promotion_min_size,

        int64_t compaction_min_size)

        : _promotion_size(promotion_size),

          _promotion_ratio(promotion_ratio),

          _promotion_min_size(promotion_min_size),

          _compaction_min_size(compaction_min_size) {}

int64_t CloudSizeBasedCumulativeCompactionPolicy::_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));

}

void find_longest_consecutive_empty_rowsets(std::vector<RowsetSharedPtr>* result,

                                            const std::vector<RowsetSharedPtr>& candidate_rowsets) {

    std::vector<RowsetSharedPtr> current_sequence;

    std::vector<RowsetSharedPtr> longest_sequence;

    for (size_t i = 0; i < candidate_rowsets.size(); ++i) {

        auto& rowset = candidate_rowsets[i];

        // Check if rowset is empty and has no delete predicate

        if (rowset->num_segments() == 0 && !rowset->rowset_meta()->has_delete_predicate()) {

            // Check if this is consecutive with previous rowset

            if (current_sequence.empty() ||

                (current_sequence.back()->end_version() == rowset->start_version() - 1)) {

                current_sequence.push_back(rowset);

            } else {

                // Start new sequence if not consecutive

                if (current_sequence.size() > longest_sequence.size()) {

                    longest_sequence = current_sequence;

                }

                current_sequence.clear();

                current_sequence.push_back(rowset);

            }

        } else {

            // Non-empty rowset, check if we have a sequence to compare

            if (current_sequence.size() > longest_sequence.size()) {

                longest_sequence = current_sequence;

            }

            current_sequence.clear();

        }

    }

    // Check final sequence

    if (current_sequence.size() > longest_sequence.size()) {

        longest_sequence = current_sequence;

    }

    *result = longest_sequence;

}

int64_t CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets(

        CloudTablet* 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(

            "CloudSizeBasedCumulativeCompactionPolicy::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);

                        }

                    }

                    LOG_INFO(

                            "[CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets.set_"

                            "input_rowsets] tablet_id={}, start={}, end={}, "

                            "input_rowsets->size()={}",

                            target_tablet_id, start_version, end_version, input_rowsets->size());

                    return input_rowsets->size();

                }

            })

    size_t promotion_size = cloud_promotion_size(tablet);

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

    int transient_size = 0;

    *compaction_score = 0;

    int64_t total_size = 0;

    bool skip_trim = false; // Skip trim for Empty Rowset Compaction

    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({

        if (skip_trim) {

            return;

        }

        // 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);

    }

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

    }

    // Check if empty rowset compaction strategy is enabled

    if (config::enable_empty_rowset_compaction && !input_rowsets->empty()) {

        // Check if input_rowsets contain consecutive empty rowsets that meet criteria

        std::vector<RowsetSharedPtr> consecutive_empty_rowsets;

        find_longest_consecutive_empty_rowsets(&consecutive_empty_rowsets, *input_rowsets);

        if (!consecutive_empty_rowsets.empty() &&

            consecutive_empty_rowsets.size() >= config::empty_rowset_compaction_min_count &&

            static_cast<double>(consecutive_empty_rowsets.size()) /

                            static_cast<double>(input_rowsets->size()) >=

                    config::empty_rowset_compaction_min_ratio) {

            // Prioritize consecutive empty rowset compaction

            // Skip trim: empty rowset compaction has very low cost and the goal is to reduce rowset count

            *input_rowsets = consecutive_empty_rowsets;

            *compaction_score = consecutive_empty_rowsets.size();

            skip_trim = true;

            return consecutive_empty_rowsets.size();

        }

    }

    auto rs_begin = input_rowsets->begin();

    size_t new_compaction_score = *compaction_score;

    const bool can_handle_exhausted_input =

            (config::prioritize_query_perf_in_compaction && tablet->keys_type() != DUP_KEYS) ||

            *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()) { // No suitable level size found in `input_rowsets`

        if (config::prioritize_query_perf_in_compaction && tablet->keys_type() != DUP_KEYS) {

            // While tablet's key type is not `DUP_KEYS`, compacting rowset in such tablets has a significant

            // positive impact on queries and reduces space amplification, so we ignore level limitation and

            // pick candidate rowsets as input rowsets.

            return transient_size;

        } else if (*compaction_score >= max_compaction_score) {

            // 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 CloudSizeBasedCumulativeCompactionPolicy::cloud_promotion_size(CloudTablet* t) const {

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

    // promotion_size is between _size_based_promotion_size and _size_based_promotion_min_size

    return promotion_size > _promotion_size       ? _promotion_size

           : promotion_size < _promotion_min_size ? _promotion_min_size

                                                  : promotion_size;

}

int64_t CloudSizeBasedCumulativeCompactionPolicy::new_cumulative_point(

        CloudTablet* tablet, const RowsetSharedPtr& output_rowset, Version& last_delete_version,

        int64_t last_cumulative_point) {

    TEST_INJECTION_POINT_RETURN_WITH_VALUE("new_cumulative_point", int64_t(0), output_rowset.get(),

                                           last_cumulative_point);

    DBUG_EXECUTE_IF("CloudSizeBasedCumulativeCompactionPolicy::new_cumulative_point", {

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

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

        if (target_tablet_id == tablet->tablet_id() && cumu_point != -1) {

            LOG_INFO(

                    "[CloudSizeBasedCumulativeCompactionPolicy::new_cumulative_point] "

                    "tablet_id={}, cumu_point={}",

                    target_tablet_id, cumu_point);

            return cumu_point;

        }

    });

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

    bool satisfy_promotion_version = tablet->enable_unique_key_merge_on_write() &&

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

                                             config::compaction_promotion_version_count;

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

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

    return (last_delete_version.first != -1 ||

            output_rowset->total_disk_size() >= cloud_promotion_size(tablet) ||

            satisfy_promotion_version)

                   ? output_rowset->end_version() + 1

                   : last_cumulative_point;

}

int64_t CloudTimeSeriesCumulativeCompactionPolicy::pick_input_rowsets(

        CloudTablet* 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) {

    int64_t last_cumu = tablet->last_cumu_compaction_success_time();

    return TimeSeriesCumulativeCompactionPolicy::pick_input_rowsets(

            tablet, last_cumu, candidate_rowsets, max_compaction_score, min_compaction_score,

            input_rowsets, last_delete_version, compaction_score, allow_delete);

}

int64_t CloudTimeSeriesCumulativeCompactionPolicy::get_compaction_level(

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

        RowsetSharedPtr output_rowset) {

    return TimeSeriesCumulativeCompactionPolicy::get_compaction_level((BaseTablet*)tablet,

                                                                      input_rowsets, output_rowset);

}

int64_t CloudTimeSeriesCumulativeCompactionPolicy::new_cumulative_point(

        CloudTablet* tablet, const RowsetSharedPtr& output_rowset, Version& last_delete_version,

        int64_t last_cumulative_point) {

    if (tablet->tablet_state() != TABLET_RUNNING || output_rowset->num_segments() == 0) {

        return last_cumulative_point;

    }

    if (tablet->tablet_meta()->time_series_compaction_level_threshold() >= 2 &&

        output_rowset->rowset_meta()->compaction_level() < 2) {

        return last_cumulative_point;

    }

    return output_rowset->end_version() + 1;

}

} // namespace doris