luwei16 commented on code in PR #65810: URL: https://github.com/apache/doris/pull/65810#discussion_r3695299450
########## be/src/storage/compaction/cumulative_compaction_binlog_policy.cpp: ########## @@ -0,0 +1,335 @@ +// 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_binlog_policy.h" + +#include <algorithm> +#include <string> + +#include "common/config.h" +#include "common/logging.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/time.h" + +namespace doris { + +bool BinlogCumulativeCompactionPolicy::is_compaction_enough( + const RowsetMetaSharedPtr& rowset_meta) const { + if (rowset_meta->compaction_level() != kBinlogCompactionMaxLevel - 1) { + return false; + } + if (rowset_meta->start_version() == 0) { + return true; + } + return rowset_meta->data_disk_size() >= + config::binlog_compaction_goal_size_mbytes * 1024 * 1024 || + rowset_meta->get_compaction_score() >= config::binlog_compaction_file_count_threshold; +} + +void BinlogCumulativeCompactionPolicy::calculate_cumulative_point( + Tablet* tablet, const RowsetMetaMapContainer& all_rowsets, int64_t current_cumulative_point, + int64_t* cumulative_point) { + *cumulative_point = Tablet::K_INVALID_CUMULATIVE_POINT; + if (current_cumulative_point != Tablet::K_INVALID_CUMULATIVE_POINT || all_rowsets.empty()) { + return; + } + + std::vector<RowsetMetaSharedPtr> rowset_metas; + rowset_metas.reserve(all_rowsets.size()); + for (const auto& [_, rs_meta] : all_rowsets) { + if (rs_meta->is_local()) { + rowset_metas.emplace_back(rs_meta); + } + } + std::sort(rowset_metas.begin(), rowset_metas.end(), RowsetMeta::comparator); + + int64_t prev_version = -1; + for (const auto& rs_meta : rowset_metas) { + if (*cumulative_point == Tablet::K_INVALID_CUMULATIVE_POINT) { + *cumulative_point = rs_meta->start_version(); + } + if (rs_meta->start_version() > prev_version + 1 || !is_compaction_enough(rs_meta)) { + break; + } + prev_version = rs_meta->end_version(); + *cumulative_point = prev_version + 1; + } + + VLOG_NOTICE << "binlog compaction policy, calculate cumulative point value=" + << *cumulative_point << ", tablet=" << tablet->tablet_id(); +} + +void BinlogCumulativeCompactionPolicy::update_cumulative_point( + Tablet* tablet, const std::vector<RowsetSharedPtr>& input_rowsets, + RowsetSharedPtr output_rowset, Version& last_delete_version) { + if (tablet->tablet_state() != TABLET_RUNNING || input_rowsets.empty() || + output_rowset->num_segments() == 0 || + output_rowset->rowset_meta()->compaction_level() != kBinlogCompactionMaxLevel - 1 || + !is_compaction_enough(output_rowset->rowset_meta())) { + return; + } + + const int64_t point = tablet->cumulative_layer_point(); + if (point == Tablet::K_INVALID_CUMULATIVE_POINT || output_rowset->start_version() > point) { + return; + } + tablet->set_cumulative_layer_point(output_rowset->end_version() + 1); +} + +int BinlogCumulativeCompactionPolicy::pick_input_rowsets( + Tablet* tablet, const std::vector<RowsetSharedPtr>& candidate_rowsets, + int8_t compaction_level, int64_t max_compaction_score, + 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 `compact_enough_rowsets` and `remaining_rowsets`. + // - L0/L1: only physical rewrite. + // - LMax: rowsets before cumulative point are candidates for quick compact. + std::vector<RowsetSharedPtr> compact_enough_rowsets; + std::vector<RowsetSharedPtr> remaining_rowsets; + compact_enough_rowsets.reserve(level_rowsets.size()); + remaining_rowsets.reserve(level_rowsets.size()); + + int compact_enough_size = 0; + const int64_t point = tablet->cumulative_layer_point(); + if (compaction_level == kBinlogCompactionMaxLevel - 1 && + point != Tablet::K_INVALID_CUMULATIVE_POINT) { + for (const auto& rs : level_rowsets) { + if (rs->end_version() < point) { + compact_enough_rowsets.push_back(rs); + ++compact_enough_size; + continue; + } + remaining_rowsets.push_back(rs); + } + } 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; + for (const auto& rs : remaining_rowsets) { + int64_t rs_score = rs->rowset_meta()->get_compaction_score(); + if (transient_size >= max_compaction_score) { + 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; Review Comment: The companion Row-Binlog Tablet never applies `binlog.ttl_seconds` (or `max_bytes` / `max_history_nums`) to its rowsets. Binlog compaction only merges historical rows, so data is retained indefinitely and storage can grow without bound. Please add retention-aware compaction/GC for both local and cloud paths. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
