be/src/cloud/cloud_cumulative_compaction_policy.cpp

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

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