[feat](row binlog)：support cloud binlog and independent binlog tablet

be/src/agent/task_worker_pool.cpp

@@ -72,6 +72,8 @@
 #include "runtime/snapshot_loader.h"
 #include "runtime/user_function_cache.h"
 #include "service/backend_options.h"
+#include "storage/compaction/cumulative_compaction_binlog_policy.h"
+#include "storage/compaction/cumulative_compaction_policy.h"
 #include "storage/compaction/cumulative_compaction_time_series_policy.h"
 #include "storage/data_dir.h"
 #include "storage/olap_common.h"
@@ -911,10 +913,11 @@ void update_tablet_meta_callback(StorageEngine& engine, const TAgentTaskRequest&
         }
         if (tablet_meta_info.__isset.compaction_policy) {
             if (tablet_meta_info.compaction_policy != CUMULATIVE_SIZE_BASED_POLICY &&
-                tablet_meta_info.compaction_policy != CUMULATIVE_TIME_SERIES_POLICY) {
+                tablet_meta_info.compaction_policy != CUMULATIVE_TIME_SERIES_POLICY &&
+                tablet_meta_info.compaction_policy != CUMULATIVE_BINLOG_POLICY) {
                 status = Status::InvalidArgument(
                         "invalid compaction policy, only support for size_based or "
-                        "time_series");
+                        "time_series or binlog");
                 continue;
             }
             tablet->tablet_meta()->set_compaction_policy(tablet_meta_info.compaction_policy);
@@ -1604,8 +1607,6 @@ void submit_table_compaction_callback(StorageEngine& engine, const TAgentTaskReq
         compaction_type = CompactionType::CUMULATIVE_COMPACTION;
     } else if (compaction_req.type == "full") {
         compaction_type = CompactionType::FULL_COMPACTION;
-    } else if (compaction_req.type == "binlog") {
-        compaction_type = CompactionType::BINLOG_COMPACTION;
     } else {
         LOG(WARNING) << "unknown compaction type: " << compaction_req.type
                      << ", tablet_id=" << compaction_req.tablet_id;
@@ -1633,17 +1634,6 @@ void submit_table_compaction_callback(StorageEngine& engine, const TAgentTaskReq
         return;
     }
 
-    if (compaction_type == CompactionType::BINLOG_COMPACTION) {
-        Status status = engine.submit_compaction_task(tablet_ptr, CompactionType::BINLOG_COMPACTION,
-                                                      /*force=*/false, /*eager=*/true,
-                                                      /*trigger_method=*/1);
-        if (!status.ok()) {
-            LOG(WARNING) << "failed to submit binlog compaction task. tablet_id="
-                         << tablet_ptr->tablet_id() << ", error=" << status;
-        }
-        return;
-    }
-
     // base / cumulative
     auto* data_dir = tablet_ptr->data_dir();
     if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) {

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -537,7 +537,8 @@ Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
     _engine.cumu_compaction_policy(compaction_policy)
             ->pick_input_rowsets(cloud_tablet(), candidate_rowsets, max_score,
                                  config::cumulative_compaction_min_deltas, &_input_rowsets,
-                                 &_last_delete_version, &compaction_score);
+                                 &_last_delete_version, &compaction_score,
+                                 _allow_delete_in_cumu_compaction);
 
     if (_input_rowsets.empty()) {
         return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(

be/src/cloud/cloud_cumulative_compaction.h

@@ -52,14 +52,14 @@ class CloudCumulativeCompaction : public CloudCompactionMixin {
 
     std::string_view compaction_name() const override { return "CloudCumulativeCompaction"; }
 
+    ReaderType compaction_type() const override { return ReaderType::READER_CUMULATIVE_COMPACTION; }
+
     Status modify_rowsets() override;
 
     Status garbage_collection() override;
 
     void update_cumulative_point();
 
-    ReaderType compaction_type() const override { return ReaderType::READER_CUMULATIVE_COMPACTION; }
-
     int64_t _input_segments = 0;
     int64_t _max_conflict_version = 0;
     // Snapshot values when pick input rowsets

be/src/cloud/cloud_cumulative_compaction_binlog_policy.cpp

@@ -0,0 +1,214 @@
+// 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_binlog_policy.h"
+
+#include "cloud/cloud_tablet.h"
+#include "common/config.h"
+#include "storage/compaction/cumulative_compaction_binlog_policy.h"
+#include "storage/tablet/tablet.h"
+#include "util/time.h"
+
+namespace doris {
+
+bool CloudBinlogCumulativeCompactionPolicy::is_compaction_enough(
+        const RowsetMetaSharedPtr& rowset_meta) const {
+    if (rowset_meta->compaction_level() !=
+        BinlogCumulativeCompactionPolicy::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;
+}
+
+int64_t CloudBinlogCumulativeCompactionPolicy::new_cumulative_point(
+        CloudTablet* tablet, const RowsetSharedPtr& output_rowset, Version& last_delete_version,
+        int64_t last_cumulative_point) {
+    if (output_rowset->num_segments() == 0 || !is_compaction_enough(output_rowset->rowset_meta())) {
+        return last_cumulative_point;
+    }
+    return output_rowset->end_version() + 1;
+}
+
+int64_t CloudBinlogCumulativeCompactionPolicy::get_compaction_level(
+        CloudTablet* tablet, const std::vector<RowsetSharedPtr>& input_rowsets,
+        RowsetSharedPtr output_rowset) {
+    DCHECK(!input_rowsets.empty()) << "tablet=" << tablet->tablet_id();
+    int64_t first_level = input_rowsets.front()->rowset_meta()->compaction_level();
+    for (size_t i = 1; i < input_rowsets.size(); ++i) {
+        DCHECK_EQ(first_level, input_rowsets[i]->rowset_meta()->compaction_level())
+                << "tablet=" << tablet->tablet_id();
+        DCHECK_EQ(input_rowsets[i]->start_version(), input_rowsets[i - 1]->end_version() + 1)
+                << "tablet=" << tablet->tablet_id();
+    }
+
+    if (first_level == BinlogCumulativeCompactionPolicy::kBinlogCompactionMaxLevel - 1) {
+        return first_level;
+    }
+    return first_level + 1;
+}
+
+uint32_t CloudBinlogCumulativeCompactionPolicy::calc_binlog_compaction_level_score(
+        CloudTablet* tablet, const std::vector<RowsetSharedPtr>& candidate_rowsets,
+        int8_t level) const {
+    uint32_t score = 0;
+    const int64_t point = tablet->cumulative_layer_point();
+    for (const auto& rs : candidate_rowsets) {
+        auto rs_meta = rs->rowset_meta();
+        if (rs_meta->compaction_level() != level) {
+            continue;
+        }
+        if (level == BinlogCumulativeCompactionPolicy::kBinlogCompactionMaxLevel - 1 &&
+            point != Tablet::K_INVALID_CUMULATIVE_POINT && rs_meta->end_version() < point) {
+            score += 1;
+            continue;
+        }
+        score += rs_meta->get_compaction_score();
+    }
+    return score;
+}
+
+uint32_t CloudBinlogCumulativeCompactionPolicy::calc_binlog_compaction_score(
+        CloudTablet* tablet, const std::vector<RowsetSharedPtr>& candidate_rowsets,
+        int8_t* compaction_level) const {
+    uint32_t max_score = 0;
+    int8_t max_level = -1;
+    for (int8_t level = 0; level < BinlogCumulativeCompactionPolicy::kBinlogCompactionMaxLevel;
+         ++level) {
+        uint32_t score = calc_binlog_compaction_level_score(tablet, candidate_rowsets, level);
+        if (score > max_score) {
+            max_score = score;
+            max_level = level;
+        }
+    }
+    if (compaction_level != nullptr) {
+        *compaction_level = max_level;
+    }
+    return max_score;
+}
+
+void CloudBinlogCumulativeCompactionPolicy::filter_new_visible_rowsets(
+        const std::vector<RowsetSharedPtr>& candidate_rowsets,
+        std::vector<RowsetSharedPtr>* output_rowsets) const {
+    output_rowsets->clear();
+    int64_t now = UnixSeconds();
+    int64_t max_processable_old_version = 0;
+    bool filter_new_rowset =
+            candidate_rowsets.size() <= config::binlog_compaction_file_count_threshold;
+    for (const auto& rs : candidate_rowsets) {
+        if (filter_new_rowset && rs->rowset_meta()->is_singleton_delta() &&
+            rs->rowset_meta()->newest_write_timestamp() +
+                            config::binlog_compaction_wait_timesec_after_visible >
+                    now) {
+            continue;
+        }
+        max_processable_old_version = std::max(max_processable_old_version, rs->end_version());
+    }
+
+    output_rowsets->reserve(candidate_rowsets.size());
+    for (const auto& rs : candidate_rowsets) {
+        if (rs->start_version() <= max_processable_old_version) {
+            output_rowsets->push_back(rs);
+        }
+    }
+}
+
+int64_t CloudBinlogCumulativeCompactionPolicy::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) {
+    std::vector<RowsetSharedPtr> filtered_rowsets;
+    filter_new_visible_rowsets(candidate_rowsets, &filtered_rowsets);
+
+    int8_t compaction_level = -1;
+    calc_binlog_compaction_score(tablet, filtered_rowsets, &compaction_level);
+    if (compaction_level < 0) {
+        *compaction_score = 0;
+        return 0;
+    }
+
+    std::vector<RowsetSharedPtr> level_rowsets;
+    level_rowsets.reserve(filtered_rowsets.size());
+    for (const auto& rs : filtered_rowsets) {
+        if (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=" << static_cast<int>(compaction_level)
+                         << ", prev_version=" << level_rowsets.back()->version()
+                         << ", next_version=" << rs->version();
+            *compaction_score = 0;
+            return 0;
+        }
+        level_rowsets.push_back(rs);
+    }
+
+    std::vector<RowsetSharedPtr> remaining_rowsets;
+    remaining_rowsets.reserve(level_rowsets.size());
+    const int64_t point = tablet->cumulative_layer_point();
+    for (const auto& rs : level_rowsets) {
+        if (compaction_level == BinlogCumulativeCompactionPolicy::kBinlogCompactionMaxLevel - 1 &&
+            point != Tablet::K_INVALID_CUMULATIVE_POINT && rs->end_version() < point) {
+            continue;
+        }
+        remaining_rowsets.push_back(rs);
+    }
+
+    std::vector<RowsetSharedPtr> picked_rowsets;
+    picked_rowsets.reserve(remaining_rowsets.size());
+    int transient_size = 0;
+    int64_t total_size = 0;
+    int64_t picked_score = 0;
+    for (const auto& rs : remaining_rowsets) {
+        if (transient_size >= max_compaction_score) {
+            break;
+        }
+        picked_rowsets.push_back(rs);
+        ++transient_size;
+        total_size += rs->data_disk_size();
+        picked_score += rs->rowset_meta()->get_compaction_score();
+    }
+
+    bool can_do_binlog_compaction =
+            total_size >= config::binlog_compaction_goal_size_mbytes * 1024 * 1024 ||
+            picked_score >= config::binlog_compaction_file_count_threshold ||
+            (UnixMillis() - tablet->last_cumu_compaction_success_time()) / 1000 >=
+                    config::binlog_compaction_time_threshold_seconds;
+    if (transient_size < 2) {
+        can_do_binlog_compaction = false;
+    }
+    if (can_do_binlog_compaction) {
+        input_rowsets->swap(picked_rowsets);
+        *compaction_score = picked_score;
+        return transient_size;
+    }
+
+    input_rowsets->clear();
+    *compaction_score = 0;
+    return 0;
+}
+
+} // namespace doris

be/src/cloud/cloud_cumulative_compaction_binlog_policy.h

@@ -0,0 +1,60 @@
+// 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.
+
+#pragma once
+
+#include "cloud/cloud_cumulative_compaction_policy.h"
+#include "storage/compaction/cumulative_compaction_binlog_policy.h"
+
+namespace doris {
+
+class CloudBinlogCumulativeCompactionPolicy final : public CloudCumulativeCompactionPolicy {
+public:
+    CloudBinlogCumulativeCompactionPolicy() = default;
+    ~CloudBinlogCumulativeCompactionPolicy() override = default;
+
+    int64_t new_cumulative_point(CloudTablet* tablet, const RowsetSharedPtr& output_rowset,
+                                 Version& last_delete_version,
+                                 int64_t last_cumulative_point) override;
+
+    int64_t get_compaction_level(CloudTablet* tablet,
+                                 const std::vector<RowsetSharedPtr>& input_rowsets,
+                                 RowsetSharedPtr output_rowset) override;
+
+    int64_t 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 = false) override;
+
+    std::string name() override { return std::string(CUMULATIVE_BINLOG_POLICY); }
+
+private:
+    bool is_compaction_enough(const RowsetMetaSharedPtr& rowset_meta) const;
+    void filter_new_visible_rowsets(const std::vector<RowsetSharedPtr>& candidate_rowsets,
+                                    std::vector<RowsetSharedPtr>* output_rowsets) const;
+    uint32_t calc_binlog_compaction_level_score(
+            CloudTablet* tablet, const std::vector<RowsetSharedPtr>& candidate_rowsets,
+            int8_t level) const;
+    uint32_t calc_binlog_compaction_score(CloudTablet* tablet,
+                                          const std::vector<RowsetSharedPtr>& candidate_rowsets,
+                                          int8_t* compaction_level) const;
+};
+
+} // namespace doris

be/src/cloud/cloud_delete_task.cpp

@@ -122,7 +122,7 @@ Status CloudDeleteTask::execute(CloudStorageEngine& engine, const TPushReq& requ
                 request.transaction_id, tablet->tablet_id(), delete_bitmap, rowset_ids, rowset,
                 request.timeout, nullptr);
     } else {
-        if (config::enable_cloud_make_rs_visible_on_be) {
+        if (config::enable_cloud_make_rs_visible_on_be && !tablet->tablet_schema()->enable_tso()) {
             engine.meta_mgr().cache_committed_rowset(rowset->rowset_meta(), context.txn_expiration);
         }
     }