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

#include <string>

#include "common/config.h"

#include "common/logging.h"

#include "storage/rowset/rowset.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_meta.h"

#include "util/time.h"

namespace doris {

int BinlogCompactionPolicy::pick_input_rowsets(

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

        int8_t compaction_level, std::vector<RowsetSharedPtr>* input_rowsets) const {

    // 1) Filter rowsets by `compaction_level`

    std::vector<RowsetSharedPtr> level_rowsets;

    level_rowsets.reserve(candidate_rowsets.size());

    for (const auto& rs : candidate_rowsets) {

        if (!rs->is_local() || rs->rowset_meta()->compaction_level() != compaction_level) {

            continue;

        }

        if (!level_rowsets.empty() &&

            rs->start_version() != level_rowsets.back()->end_version() + 1) {

            LOG(WARNING) << "rowset is non-continuous in the same compaction_level of binlog "

                            "compaction. tablet="

                         << tablet->tablet_id()

                         << ", compaction_level=" << std::to_string(compaction_level)

                         << ", prev_version=" << level_rowsets.back()->version()

                         << ", next_version=" << rs->version();

            return 0;

        }

        level_rowsets.push_back(rs);

    }

    // 2) Split `level_rowsets` into `full_enough_rowsets` and `remaining_rowsets`.

    //    - L0/L1: only physical rewrite.

    //    - LMax: Base([0-x]) + prefix ENOUGH rowsets are candidates for quick compact.

    std::vector<RowsetSharedPtr> full_enough_rowsets;

    std::vector<RowsetSharedPtr> remaining_rowsets;

    full_enough_rowsets.reserve(level_rowsets.size());

    remaining_rowsets.reserve(level_rowsets.size());

    bool find_base = false;

    int full_enough_size = 0;

    size_t idx = 0;

    if (compaction_level == kBinlogCompactionMaxLevel - 1) {

        if (!level_rowsets.empty() && level_rowsets[0]->start_version() == 0) {

            find_base = true;

            full_enough_rowsets.push_back(level_rowsets[0]);

            ++full_enough_size;

            idx = 1;

            for (; idx < level_rowsets.size(); ++idx) {

                const auto& rs = level_rowsets[idx];

                int64_t rs_score = rs->rowset_meta()->get_compaction_score();

                if (rs->data_disk_size() >=

                            config::binlog_compaction_goal_size_mbytes * 1024 * 1024 ||

                    rs_score >= config::binlog_compaction_file_count_threshold) {

                    full_enough_rowsets.push_back(rs);

                    ++full_enough_size;

                    continue;

                }

                break;

            }

            for (; idx < level_rowsets.size(); ++idx) {

                remaining_rowsets.push_back(level_rowsets[idx]);

            }

        } else {

            remaining_rowsets = level_rowsets;

        }

    } else {

        remaining_rowsets = level_rowsets;

    }

    // 3) Pick rowsets from `remaining_rowsets` (physical rewrite path).

    std::vector<RowsetSharedPtr> picked_rowsets;

    picked_rowsets.reserve(remaining_rowsets.size());

    int transient_size = 0;

    int64_t total_size = 0;

    int64_t compaction_score = 0;

    const int64_t max_binlog_permits = (config::total_permits_for_compaction_score *

                                                config::binlog_compaction_permits_percent +

                                        99) /

                                       100;

    for (const auto& rs : remaining_rowsets) {

        int64_t rs_score = rs->rowset_meta()->get_compaction_score();

        if (transient_size >= config::binlog_level_compaction_max_deltas ||

            compaction_score + rs_score > max_binlog_permits) {

            break;

        }

        picked_rowsets.push_back(rs);

        ++transient_size;

        total_size += rs->data_disk_size();

        compaction_score += rs_score;

    }

    // 4) Trigger check

    //    - L0/L1: only physical rewrite if trigger is met.

    //    - LMax: if physical rewrite trigger is NOT met, try quick compact; if both met, compare score.

    bool can_do_binlog_compaction = false;

    if (total_size >= config::binlog_compaction_goal_size_mbytes * 1024 * 1024 ||

        compaction_score >= config::binlog_compaction_file_count_threshold) {

        can_do_binlog_compaction = true;

    } else if ((UnixMillis() - tablet->last_binlog_compaction_success_time(compaction_level)) /

                       1000 >=

               config::binlog_compaction_time_threshold_seconds) {

        can_do_binlog_compaction = true;

    }

    // 5) LMax quick compact vs physical rewrite.

    if (compaction_level == kBinlogCompactionMaxLevel - 1 && find_base && full_enough_size > 1) {

        int64_t quick_merge_score = 1;

        for (int i = 1; i < full_enough_size; ++i) {

            quick_merge_score += full_enough_rowsets[i]->rowset_meta()->get_compaction_score();

        }

        if (!can_do_binlog_compaction || quick_merge_score > compaction_score) {

            input_rowsets->swap(full_enough_rowsets);

            return full_enough_size;

        }

    }

    if (transient_size < 2) {

        can_do_binlog_compaction = false;

    }

    if (can_do_binlog_compaction) {

        input_rowsets->swap(picked_rowsets);

        return transient_size;

    }

    input_rowsets->clear();

    return 0;

}

uint32_t BinlogCompactionPolicy::calc_binlog_compaction_level_score(Tablet* tablet,

                                                                    int8_t level) const {

    uint32_t score = 0;

    // Binlog tiered compaction score (L0..LMax)

    //

    //   L0/L1/... : | rowsets ... |

    //   LMax      : | Base([0-x]) | remaining rowsets ... |

    //

    // Base only performs meta-only merge (merge rowset meta), so get_compaction_score()

    // treats Base score as 1.

    for (const auto& [_, rs_meta] : tablet->tablet_meta()->all_row_binlog_rs_metas()) {

        if (!rs_meta->is_local() || rs_meta->compaction_level() != level) {

            continue;

        }

        score += rs_meta->get_compaction_score();

    }

    return score;

}

uint32_t BinlogCompactionPolicy::calc_binlog_compaction_score(

        Tablet* tablet, int8_t* prefer_compaction_level) const {

    uint32_t max_score = 0;

    int8_t max_level = -1;

    for (int8_t level = 0; level < kBinlogCompactionMaxLevel; ++level) {

        uint32_t score = calc_binlog_compaction_level_score(tablet, level);

        if (score > max_score) {

            max_score = score;

            max_level = level;

        }

    }

    if (prefer_compaction_level != nullptr) {

        *prefer_compaction_level = max_level;

    }

    return max_score;

}

void BinlogCompactionPolicy::update_compaction_level(

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

        RowsetSharedPtr output_rowset) {

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

        return;

    }

    int64_t first_level = 0;

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

        int64_t cur_level = input_rowsets[i]->rowset_meta()->compaction_level();

        if (i == 0) {

            first_level = cur_level;

            continue;

        }

        DCHECK_EQ(first_level, cur_level)

                << "Compaction level mismatch, first_level: " << first_level

                << ", cur_level: " << cur_level << ", tablet: " << tablet->tablet_id();

        if (first_level != cur_level) {

            LOG(WARNING) << "Failed to check compaction level, first_level: " << first_level

                         << ", cur_level: " << cur_level << ", tablet: " << tablet->tablet_id();

        }

        DCHECK_EQ(input_rowsets[i]->start_version(), input_rowsets[i - 1]->end_version() + 1)

                << "Binlog compaction input rowsets are non-continuous, prev_version: "

                << input_rowsets[i - 1]->version()

                << ", cur_version: " << input_rowsets[i]->version()

                << ", tablet: " << tablet->tablet_id();

        if (input_rowsets[i]->start_version() != input_rowsets[i - 1]->end_version() + 1) {

            LOG(WARNING) << "Failed to check binlog compaction input rowsets continuity, "

                         << "prev_version=" << input_rowsets[i - 1]->version()

                         << ", cur_version=" << input_rowsets[i]->version()

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

        }

    }

    if (first_level == kBinlogCompactionMaxLevel - 1) {

        // level max do not update compaction level

        output_rowset->rowset_meta()->set_compaction_level(first_level);

        return;

    }

    output_rowset->rowset_meta()->set_compaction_level(first_level + 1);

}

} // namespace doris