be/src/storage/rowset/beta_rowset_writer.cpp

Source
// 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/rowset/beta_rowset_writer.h"

#include <assert.h>
// IWYU pragma: no_include <bthread/errno.h>
#include <errno.h> // IWYU pragma: keep
#include <fmt/format.h>
#include <stdio.h>

#include <chrono>
#include <ctime> // time
#include <filesystem>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>
#include <utility>
#include <vector>

// IWYU pragma: no_include <opentelemetry/common/threadlocal.h>
#include "common/cast_set.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/column/column.h"
#include "core/data_type/data_type_factory.hpp"
#include "io/fs/file_reader.h"
#include "io/fs/file_system.h"
#include "io/fs/file_writer.h"
#include "runtime/thread_context.h"
#include "storage/index/inverted/inverted_index_cache.h"
#include "storage/index/inverted/inverted_index_desc.h"
#include "storage/olap_define.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/rowset/rowset_factory.h"
#include "storage/rowset/rowset_writer.h"
#include "storage/rowset/segcompaction.h"
#include "storage/schema_change/schema_change.h"
#include "storage/segment/segment.h"
#include "storage/segment/segment_writer.h"
#include "storage/storage_engine.h"
#include "storage/tablet/tablet_schema.h"
#include "util/debug_points.h"
#include "util/pretty_printer.h"
#include "util/slice.h"
#include "util/stopwatch.hpp"
#include "util/time.h"

namespace doris {
using namespace ErrorCode;

namespace {

bool is_segment_overlapping(const std::vector<KeyBoundsPB>& segments_encoded_key_bounds) {
    std::string_view last;
    for (auto&& segment_encode_key : segments_encoded_key_bounds) {
        auto&& cur_min = segment_encode_key.min_key();
        auto&& cur_max = segment_encode_key.max_key();
        if (cur_min <= last) {
            return true;
        }
        last = cur_max;
    }
    return false;
}

bool copy_key_bounds_with_truncation(const KeyBoundsPB& src, KeyBoundsPB* dst) {
    DCHECK(dst != nullptr);
    if (config::random_segments_key_bounds_truncation) {
        dst->CopyFrom(src);
        return false;
    }
    const int32_t truncation_threshold = config::segments_key_bounds_truncation_threshold;
    if (truncation_threshold <= 0) {
        dst->CopyFrom(src);
        return false;
    }
    const size_t truncation_size = cast_set<size_t>(truncation_threshold);

    bool truncated = false;
    auto copy_key = [&](const std::string& key, std::string* stored_key) {
        if (key.size() > truncation_size) {
            stored_key->assign(key.data(), truncation_size);
            truncated = true;
            return;
        }
        stored_key->assign(key.data(), key.size());
    };
    copy_key(src.min_key(), dst->mutable_min_key());
    copy_key(src.max_key(), dst->mutable_max_key());
    return truncated;
}

SegmentStatistics copy_segment_statistics_with_truncated_key_bounds(const SegmentStatistics& src,
                                                                    bool& key_bounds_truncated) {
    SegmentStatistics dst;
    dst.row_num = src.row_num;
    dst.data_size = src.data_size;
    dst.index_size = src.index_size;
    key_bounds_truncated = copy_key_bounds_with_truncation(src.key_bounds, &dst.key_bounds);
    return dst;
}

void build_rowset_meta_with_spec_field(RowsetMeta& rowset_meta,
                                       const RowsetMeta& spec_rowset_meta) {
    rowset_meta.set_num_rows(spec_rowset_meta.num_rows());
    rowset_meta.set_total_disk_size(spec_rowset_meta.total_disk_size());
    rowset_meta.set_data_disk_size(spec_rowset_meta.data_disk_size());
    rowset_meta.set_index_disk_size(spec_rowset_meta.index_disk_size());
    // TODO write zonemap to meta
    rowset_meta.set_empty(spec_rowset_meta.num_rows() == 0);
    rowset_meta.set_creation_time(time(nullptr));
    rowset_meta.set_num_segments(spec_rowset_meta.num_segments());
    rowset_meta.set_segments_overlap(spec_rowset_meta.segments_overlap());
    rowset_meta.set_rowset_state(spec_rowset_meta.rowset_state());
    rowset_meta.set_segments_key_bounds_truncated(
            spec_rowset_meta.is_segments_key_bounds_truncated());
    rowset_meta.set_db_id(spec_rowset_meta.db_id());
    rowset_meta.set_table_id(spec_rowset_meta.table_id());
    std::vector<KeyBoundsPB> segments_key_bounds;
    spec_rowset_meta.get_segments_key_bounds(&segments_key_bounds);
    // Preserve source layout: if source was aggregated (size 1), re-aggregating
    // the single entry is a no-op that also keeps the flag consistent.
    rowset_meta.set_segments_key_bounds(segments_key_bounds,
                                        spec_rowset_meta.is_segments_key_bounds_aggregated());
    std::vector<uint32_t> num_segment_rows;
    spec_rowset_meta.get_num_segment_rows(&num_segment_rows);
    rowset_meta.set_num_segment_rows(num_segment_rows);
    if (spec_rowset_meta.has_commit_tso()) {
        rowset_meta.set_commit_tso(spec_rowset_meta.commit_tso());
    }
    if (spec_rowset_meta.is_row_binlog()) {
        rowset_meta.mark_row_binlog();
    }
}

} // namespace

SegmentFileCollection::~SegmentFileCollection() = default;

Status SegmentFileCollection::add(int seg_id, io::FileWriterPtr&& writer) {
    std::lock_guard lock(_lock);
    if (_closed) [[unlikely]] {
        DCHECK(false) << writer->path();
        return Status::InternalError("add to closed SegmentFileCollection");
    }

    _file_writers.emplace(seg_id, std::move(writer));
    return Status::OK();
}

io::FileWriter* SegmentFileCollection::get(int seg_id) const {
    std::lock_guard lock(_lock);
    if (auto it = _file_writers.find(seg_id); it != _file_writers.end()) {
        return it->second.get();
    } else {
        return nullptr;
    }
}

Status SegmentFileCollection::close() {
    {
        std::lock_guard lock(_lock);
        if (_closed) [[unlikely]] {
            DCHECK(false);
            return Status::InternalError("double close SegmentFileCollection");
        }
        _closed = true;
    }

    for (auto&& [_, writer] : _file_writers) {
        DBUG_EXECUTE_IF("SegmentFileCollection.close.wait_dat_closed", {
            auto before_state = writer->state();
            for (int i = 0; i < 3000 && writer->state() != io::FileWriter::State::CLOSED; ++i) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
            LOG(INFO) << "SegmentFileCollection.close.wait_dat_closed path="
                      << writer->path().native()
                      << " before_state=" << static_cast<int>(before_state)
                      << " after_state=" << static_cast<int>(writer->state());
        });
        if (writer->state() != io::FileWriter::State::CLOSED) {
            RETURN_IF_ERROR(writer->close());
        }
    }

    return Status::OK();
}

Result<std::vector<size_t>> SegmentFileCollection::segments_file_size(int seg_id_offset) {
    std::lock_guard lock(_lock);
    if (!_closed) [[unlikely]] {
        DCHECK(false);
        return ResultError(Status::InternalError("get segments file size without closed"));
    }

    Status st;
    std::vector<size_t> seg_file_size(_file_writers.size(), 0);
    bool succ = std::all_of(_file_writers.begin(), _file_writers.end(), [&](auto&& it) {
        auto&& [seg_id, writer] = it;

        int idx = seg_id - seg_id_offset;
        if (idx >= seg_file_size.size()) [[unlikely]] {
            auto err_msg = fmt::format(
                    "invalid seg_id={} num_file_writers={} seg_id_offset={} path={}", seg_id,
                    seg_file_size.size(), seg_id_offset, writer->path().native());
            DCHECK(false) << err_msg;
            st = Status::InternalError(err_msg);
            return false;
        }

        auto& fsize = seg_file_size[idx];
        if (fsize != 0) {
            // File size should not been set
            auto err_msg =
                    fmt::format("duplicate seg_id={} path={}", seg_id, writer->path().native());
            DCHECK(false) << err_msg;
            st = Status::InternalError(err_msg);
            return false;
        }

        fsize = writer->bytes_appended();
        if (fsize <= 0) {
            auto err_msg =
                    fmt::format("invalid segment fsize={} path={}", fsize, writer->path().native());
            DCHECK(false) << err_msg;
            st = Status::InternalError(err_msg);
            return false;
        }

        return true;
    });

    if (succ) {
        return seg_file_size;
    }

    return ResultError(st);
}

InvertedIndexFileCollection::~InvertedIndexFileCollection() = default;

Status InvertedIndexFileCollection::add(int seg_id, IndexFileWriterPtr&& index_writer) {
    std::lock_guard lock(_lock);
    if (_inverted_index_file_writers.find(seg_id) != _inverted_index_file_writers.end())
            [[unlikely]] {
        DCHECK(false);
        return Status::InternalError("The seg_id already exists, seg_id is {}", seg_id);
    }
    _inverted_index_file_writers.emplace(seg_id, std::move(index_writer));
    return Status::OK();
}

Status InvertedIndexFileCollection::begin_close() {
    std::lock_guard lock(_lock);
    for (auto&& [id, writer] : _inverted_index_file_writers) {
        RETURN_IF_ERROR(writer->begin_close());
        _total_size += writer->get_index_file_total_size();
    }

    return Status::OK();
}

Status InvertedIndexFileCollection::finish_close() {
    std::lock_guard lock(_lock);
    for (auto&& [id, writer] : _inverted_index_file_writers) {
        RETURN_IF_ERROR(writer->finish_close());
    }
    return Status::OK();
}

Result<std::vector<const InvertedIndexFileInfo*>>
InvertedIndexFileCollection::inverted_index_file_info(int seg_id_offset) {
    std::lock_guard lock(_lock);

    Status st;
    std::vector<const InvertedIndexFileInfo*> idx_file_info(_inverted_index_file_writers.size());
    bool succ = std::all_of(
            _inverted_index_file_writers.begin(), _inverted_index_file_writers.end(),
            [&](auto&& it) {
                auto&& [seg_id, writer] = it;

                int idx = seg_id - seg_id_offset;
                if (idx >= idx_file_info.size()) [[unlikely]] {
                    auto err_msg =
                            fmt::format("invalid seg_id={} num_file_writers={} seg_id_offset={}",
                                        seg_id, idx_file_info.size(), seg_id_offset);
                    DCHECK(false) << err_msg;
                    st = Status::InternalError(err_msg);
                    return false;
                }
                idx_file_info[idx] = _inverted_index_file_writers[seg_id]->get_index_file_info();
                return true;
            });

    if (succ) {
        return idx_file_info;
    }

    return ResultError(st);
}

BaseBetaRowsetWriter::BaseBetaRowsetWriter()
        : _num_segment(0),
          _segment_start_id(0),
          _num_rows_written(0),
          _total_data_size(0),
          _total_index_size(0),
          _segment_creator(_context, _seg_files, _idx_files) {}

BetaRowsetWriter::BetaRowsetWriter(StorageEngine& engine)
        : _engine(engine), _segcompaction_worker(std::make_shared<SegcompactionWorker>(this)) {}

RowBinlogRowsetWriter::RowBinlogRowsetWriter(StorageEngine& engine) : BetaRowsetWriter(engine) {}

BaseBetaRowsetWriter::~BaseBetaRowsetWriter() {
    if (!_already_built && _rowset_meta->is_local()) {
        // abnormal exit, remove all files generated
        auto& fs = io::global_local_filesystem();
        for (int i = _segment_start_id; i < _segment_creator.next_segment_id(); ++i) {
            std::string seg_path =
                    local_segment_path(_context.tablet_path, _context.rowset_id.to_string(), i);
            // Even if an error is encountered, these files that have not been cleaned up
            // will be cleaned up by the GC background. So here we only print the error
            // message when we encounter an error.
            WARN_IF_ERROR(fs->delete_file(seg_path),
                          fmt::format("Failed to delete file={}", seg_path));
        }
    }
    if (_calc_delete_bitmap_token) {
        _calc_delete_bitmap_token->cancel();
    }
}

BetaRowsetWriter::~BetaRowsetWriter() {
    /* Note that segcompaction is async and in parallel with load job. So we should handle carefully
     * when the job is cancelled. Although it is meaningless to continue segcompaction when the job
     * is cancelled, the objects involved in the job should be preserved during segcompaction to
     * avoid crashs for memory issues. */
    WARN_IF_ERROR(_wait_flying_segcompaction(), "segment compaction failed");
}

Status BaseBetaRowsetWriter::init(const RowsetWriterContext& rowset_writer_context) {
    _context = rowset_writer_context;
    DCHECK(_context.tablet_schema != nullptr);
    _rowset_meta.reset(new RowsetMeta);
    if (_context.storage_resource) {
        _rowset_meta->set_remote_storage_resource(*_context.storage_resource);
    }
    _rowset_meta->set_rowset_id(_context.rowset_id);
    _rowset_meta->set_partition_id(_context.partition_id);
    _rowset_meta->set_tablet_id(_context.tablet_id);
    _rowset_meta->set_db_id(_context.db_id);
    _rowset_meta->set_table_id(_context.table_id);
    _rowset_meta->set_index_id(_context.index_id);
    _rowset_meta->set_tablet_schema_hash(_context.tablet_schema_hash);
    _rowset_meta->set_rowset_type(_context.rowset_type);
    _rowset_meta->set_rowset_state(_context.rowset_state);
    _rowset_meta->set_segments_overlap(_context.segments_overlap);
    if (_context.rowset_state == PREPARED || _context.rowset_state == COMMITTED) {
        _is_pending = true;
        _rowset_meta->set_txn_id(_context.txn_id);
        _rowset_meta->set_load_id(_context.load_id);
    } else {
        _rowset_meta->set_version(_context.version);
        _rowset_meta->set_newest_write_timestamp(_context.newest_write_timestamp);
    }
    _rowset_meta->set_tablet_uid(_context.tablet_uid);
    _rowset_meta->set_tablet_schema(_context.tablet_schema);
    _rowset_meta->set_compaction_level(_context.compaction_level);
    if (_context.write_binlog_opt().enable) {
        _rowset_meta->mark_row_binlog();
    }
    _context.segment_collector = std::make_shared<SegmentCollectorT<BaseBetaRowsetWriter>>(this);
    _context.file_writer_creator = std::make_shared<FileWriterCreatorT<BaseBetaRowsetWriter>>(this);
    return Status::OK();
}

Status BaseBetaRowsetWriter::add_block(const Block* block) {
    return _segment_creator.add_block(block);
}

bool BaseBetaRowsetWriter::_is_segment_delete_bitmap_calculated(uint32_t segment_id) const {
    if (_context.mow_context == nullptr) {
        // This segcompaction gate is only for local MoW rowsets. Non-MoW rowsets
        // have no pending async delete bitmap work.
        return true;
    }
    std::lock_guard lock(_delete_bitmap_calculated_segments_mutex);
    return _delete_bitmap_calculated_segments.contains(segment_id);
}

void BaseBetaRowsetWriter::_mark_segment_delete_bitmap_calculated(uint32_t segment_id) {
    std::lock_guard lock(_delete_bitmap_calculated_segments_mutex);
    _delete_bitmap_calculated_segments.add(segment_id);
}

Status BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) {
    SCOPED_RAW_TIMER(&_delete_bitmap_ns);
    if (_context.is_transient_rowset_writer ||
        !_context.tablet->enable_unique_key_merge_on_write() ||
        (_context.partial_update_info && _context.partial_update_info->is_partial_update())) {
        _mark_segment_delete_bitmap_calculated(segment_id);
        return Status::OK();
    }
    std::vector<RowsetSharedPtr> specified_rowsets;
    {
        std::shared_lock meta_rlock(_context.tablet->get_header_lock());
        specified_rowsets =
                _context.tablet->get_rowset_by_ids(_context.mow_context->rowset_ids.get());
    }

    // Submit the entire delete bitmap calculation process to thread pool for async execution
    // This avoids blocking memtable flush thread while waiting for file upload to complete
    // The process includes: file_writer->close(), _build_tmp, load_segments, and calc_delete_bitmap
    return _calc_delete_bitmap_token->submit_func(
            [this, segment_id, specified_rowsets = std::move(specified_rowsets)]() -> Status {
                Status st = Status::OK();
                // If this async pipeline fails, leave the segment unmarked. build() waits
                // on the token and aborts this writer; marking here would allow
                // segcompaction to consume a segment whose delete bitmap was not published.
                // Step 1: Close file_writer (must be done before load_segments)
                auto* file_writer = _seg_files.get(segment_id);
                if (file_writer && file_writer->state() != io::FileWriter::State::CLOSED) {
                    MonotonicStopWatch close_timer;
                    close_timer.start();
                    st = file_writer->close();
                    close_timer.stop();

                    auto close_time_ms = close_timer.elapsed_time_milliseconds();
                    if (close_time_ms > 1000) {
                        LOG(INFO) << "file_writer->close() took " << close_time_ms
                                  << "ms for segment_id=" << segment_id
                                  << ", tablet_id=" << _context.tablet_id
                                  << ", rowset_id=" << _context.rowset_id;
                    }
                    if (!st.ok()) {
                        return st;
                    }
                }

                DBUG_EXECUTE_IF("BetaRowsetWriter.generate_delete_bitmap.sleep_before_build_tmp", {
                    auto sleep_ms = dp->param<int64_t>("sleep_ms", 30000);
                    std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
                });

                OlapStopWatch watch;
                // Step 2: Build tmp rowset (needs file_writer to be closed)
                RowsetSharedPtr rowset_ptr;
                st = _build_tmp(rowset_ptr);
                if (!st.ok()) {
                    return st;
                }

                // Step 3: Load segments (needs file_writer to be closed and rowset to be built)
                auto* beta_rowset = reinterpret_cast<BetaRowset*>(rowset_ptr.get());
                std::vector<segment_v2::SegmentSharedPtr> segments;
                st = beta_rowset->load_segments(segment_id, segment_id + 1, &segments);
                if (!st.ok()) {
                    return st;
                }

                // Step 4: Calculate delete bitmap
                st = BaseTablet::calc_delete_bitmap(
                        _context.tablet, rowset_ptr, segments, specified_rowsets,
                        _context.mow_context->delete_bitmap, _context.mow_context->max_version,
                        nullptr, nullptr, nullptr);
                if (!st.ok()) {
                    return st;
                }

                size_t total_rows =
                        std::accumulate(segments.begin(), segments.end(), 0,
                                        [](size_t sum, const segment_v2::SegmentSharedPtr& s) {
                                            return sum += s->num_rows();
                                        });
                LOG(INFO) << "[Memtable Flush] construct delete bitmap tablet: "
                          << _context.tablet->tablet_id()
                          << ", rowset_ids: " << _context.mow_context->rowset_ids->size()
                          << ", cur max_version: " << _context.mow_context->max_version
                          << ", transaction_id: " << _context.mow_context->txn_id
                          << ", delete_bitmap_count: "
                          << _context.mow_context->delete_bitmap->get_delete_bitmap_count()
                          << ", delete_bitmap_cardinality: "
                          << _context.mow_context->delete_bitmap->cardinality()
                          << ", cost: " << watch.get_elapse_time_us()
                          << "(us), total rows: " << total_rows;
                // Mark only after delete bitmap entries are published to mow_context. Otherwise
                // segcompaction could convert bitmap before this segment's entries exist.
                _mark_segment_delete_bitmap_calculated(segment_id);
                return Status::OK();
            });
}

Status BetaRowsetWriter::init(const RowsetWriterContext& rowset_writer_context) {
    RETURN_IF_ERROR(BaseBetaRowsetWriter::init(rowset_writer_context));
    if (_segcompaction_worker) {
        _segcompaction_worker->init_mem_tracker(rowset_writer_context);
    }
    if (_context.mow_context != nullptr) {
        _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token();
    }
    return Status::OK();
}

Status BetaRowsetWriter::_load_noncompacted_segment(segment_v2::SegmentSharedPtr& segment,
                                                    int32_t segment_id) {
    DCHECK(_rowset_meta->is_local());
    auto fs = _rowset_meta->fs();
    if (!fs) {
        return Status::Error<INIT_FAILED>(
                "BetaRowsetWriter::_load_noncompacted_segment _rowset_meta->fs get failed");
    }
    auto path =
            local_segment_path(_context.tablet_path, _context.rowset_id.to_string(), segment_id);
    io::FileReaderOptions reader_options {
            .cache_type =
                    _context.write_file_cache
                            ? (config::enable_file_cache ? io::FileCachePolicy::FILE_BLOCK_CACHE
                                                         : io::FileCachePolicy::NO_CACHE)
                            : io::FileCachePolicy::NO_CACHE,
            .is_doris_table = true,
            .cache_base_path {},
            .tablet_id = _rowset_meta->tablet_id(),
    };
    auto s = segment_v2::Segment::open(fs, path, _rowset_meta->tablet_id(), segment_id, rowset_id(),
                                       _context.tablet_schema, reader_options, &segment);
    if (!s.ok()) {
        LOG(WARNING) << "failed to open segment. " << path << ":" << s;
        return s;
    }
    return Status::OK();
}

/* policy of segcompaction target selection:
 *  1. skip big segments
 *  2. if the consecutive smalls end up with a big, compact the smalls, except
 *     single small
 *  3. if the consecutive smalls end up with small, compact the smalls if the
 *     length is beyond (config::segcompaction_batch_size / 2)
 */
// NOLINTNEXTLINE(readability-function-cognitive-complexity): existing selection logic is kept intact.
Status BetaRowsetWriter::_find_longest_consecutive_small_segment(
        SegCompactionCandidatesSharedPtr& segments) {
    segments = std::make_shared<SegCompactionCandidates>();
    // skip last (maybe active) segment
    int32_t last_segment = _num_segment - 1;
    size_t task_bytes = 0;
    uint32_t task_rows = 0;
    int32_t segid;
    for (segid = _segcompacted_point;
         segid < last_segment && segments->size() < config::segcompaction_batch_size; segid++) {
        if (!_is_segment_delete_bitmap_calculated(segid)) {
            VLOG_DEBUG << "stop segcompaction at segid=" << segid
                       << " because its delete bitmap is not calculated yet, tablet_id="
                       << _context.tablet_id << ", rowset_id=" << _context.rowset_id;
            break;
        }
        segment_v2::SegmentSharedPtr segment;
        RETURN_IF_ERROR(_load_noncompacted_segment(segment, segid));
        const auto segment_rows = segment->num_rows();
        const auto segment_bytes = segment->file_reader()->size();
        bool is_large_segment = segment_rows > config::segcompaction_candidate_max_rows ||
                                segment_bytes > config::segcompaction_candidate_max_bytes;
        if (is_large_segment) {
            if (segid == _segcompacted_point) {
                // skip large segments at the front
                auto dst_seg_id = _num_segcompacted.load();
                RETURN_IF_ERROR(_rename_compacted_segment_plain(_segcompacted_point++));
                if (_segcompaction_worker->need_convert_delete_bitmap()) {
                    _segcompaction_worker->convert_segment_delete_bitmap(
                            _context.mow_context->delete_bitmap, segid, dst_seg_id);
                }
                continue;
            } else {
                // stop because we need consecutive segments
                break;
            }
        }
        bool is_task_full = task_rows + segment_rows > config::segcompaction_task_max_rows ||
                            task_bytes + segment_bytes > config::segcompaction_task_max_bytes;
        if (is_task_full) {
            break;
        }
        segments->push_back(segment);
        task_rows += segment->num_rows();
        task_bytes += segment->file_reader()->size();
    }
    size_t s = segments->size();
    const size_t min_final_batch_size = cast_set<size_t>(config::segcompaction_batch_size / 2);
    if (segid == last_segment && s <= min_final_batch_size) {
        // we didn't collect enough segments, better to do it in next
        // round to compact more at once
        segments->clear();
        return Status::OK();
    }
    if (s == 1) { // poor bachelor, let it go
        VLOG_DEBUG << "only one candidate segment";
        auto src_seg_id = _segcompacted_point.load();
        auto dst_seg_id = _num_segcompacted.load();
        RETURN_IF_ERROR(_rename_compacted_segment_plain(_segcompacted_point++));
        if (_segcompaction_worker->need_convert_delete_bitmap()) {
            _segcompaction_worker->convert_segment_delete_bitmap(
                    _context.mow_context->delete_bitmap, src_seg_id, dst_seg_id);
        }
        segments->clear();
        return Status::OK();
    }
    if (VLOG_DEBUG_IS_ON) {
        vlog_buffer.clear();
        for (auto& segment : (*segments.get())) {
            fmt::format_to(vlog_buffer, "[id:{} num_rows:{}]", segment->id(), segment->num_rows());
        }
        VLOG_DEBUG << "candidate segments num:" << s
                   << " list of candidates:" << fmt::to_string(vlog_buffer);
    }
    return Status::OK();
}

Status BetaRowsetWriter::_rename_compacted_segments(int64_t begin, int64_t end) {
    int ret;
    auto src_seg_path = BetaRowset::local_segment_path_segcompacted(_context.tablet_path,
                                                                    _context.rowset_id, begin, end);
    auto dst_seg_path = local_segment_path(_context.tablet_path, _context.rowset_id.to_string(),
                                           _num_segcompacted);
    ret = rename(src_seg_path.c_str(), dst_seg_path.c_str());
    if (ret) {
        return Status::Error<ROWSET_RENAME_FILE_FAILED>(
                "failed to rename {} to {}. ret:{}, errno:{}", src_seg_path, dst_seg_path, ret,
                errno);
    }
    RETURN_IF_ERROR(_remove_segment_footer_cache(_num_segcompacted, dst_seg_path));

    // rename inverted index files
    RETURN_IF_ERROR(_rename_compacted_indices(begin, end, 0));

    _num_segcompacted++;
    return Status::OK();
}

void BetaRowsetWriter::_clear_statistics_for_deleting_segments_unsafe(uint32_t begin,
                                                                      uint32_t end) {
    VLOG_DEBUG << "_segid_statistics_map clear record segid range from:" << begin << " to:" << end;
    for (uint32_t i = begin; i <= end; ++i) {
        _segid_statistics_map.erase(i);
    }
}

Status BetaRowsetWriter::_rename_compacted_segment_plain(uint32_t seg_id) {
    if (seg_id == cast_set<uint32_t>(_num_segcompacted.load())) {
        ++_num_segcompacted;
        return Status::OK();
    }

    auto src_seg_path =
            local_segment_path(_context.tablet_path, _context.rowset_id.to_string(), seg_id);
    auto dst_seg_path = local_segment_path(_context.tablet_path, _context.rowset_id.to_string(),
                                           _num_segcompacted);
    VLOG_DEBUG << "segcompaction skip this segment. rename " << src_seg_path << " to "
               << dst_seg_path;
    {
        std::lock_guard<std::mutex> lock(_segid_statistics_map_mutex);
        DCHECK_EQ(_segid_statistics_map.find(seg_id) == _segid_statistics_map.end(), false);
        DCHECK_EQ(_segid_statistics_map.find(_num_segcompacted) == _segid_statistics_map.end(),
                  true);
        auto org = _segid_statistics_map[seg_id];
        _segid_statistics_map.emplace(_num_segcompacted, org);
        _clear_statistics_for_deleting_segments_unsafe(seg_id, seg_id);
    }
    int ret = rename(src_seg_path.c_str(), dst_seg_path.c_str());
    if (ret) {
        return Status::Error<ROWSET_RENAME_FILE_FAILED>(
                "failed to rename {} to {}. ret:{}, errno:{}", src_seg_path, dst_seg_path, ret,
                errno);
    }

    RETURN_IF_ERROR(_remove_segment_footer_cache(_num_segcompacted, dst_seg_path));
    // rename remaining inverted index files
    RETURN_IF_ERROR(_rename_compacted_indices(-1, -1, seg_id));

    ++_num_segcompacted;
    return Status::OK();
}

Status BetaRowsetWriter::_remove_segment_footer_cache(const uint32_t seg_id,
                                                      const std::string& segment_path) {
    auto* footer_page_cache = ExecEnv::GetInstance()->get_storage_page_cache();
    if (!footer_page_cache) {
        return Status::OK();
    }

    auto fs = _rowset_meta->fs();
    bool exists = false;
    RETURN_IF_ERROR(fs->exists(segment_path, &exists));
    if (exists) {
        io::FileReaderSPtr file_reader;
        io::FileReaderOptions reader_options {
                .cache_type = config::enable_file_cache ? io::FileCachePolicy::FILE_BLOCK_CACHE
                                                        : io::FileCachePolicy::NO_CACHE,
                .is_doris_table = true,
                .cache_base_path = "",
                .file_size = _rowset_meta->segment_file_size(static_cast<int>(seg_id)),
                .tablet_id = _rowset_meta->tablet_id(),
        };
        RETURN_IF_ERROR(fs->open_file(segment_path, &file_reader, &reader_options));
        DCHECK(file_reader != nullptr);
        auto cache_key = segment_v2::Segment::get_segment_footer_cache_key(file_reader);
        footer_page_cache->erase(cache_key, segment_v2::PageTypePB::INDEX_PAGE);
    }
    return Status::OK();
}

Status BetaRowsetWriter::_rename_compacted_indices(int64_t begin, int64_t end, uint64_t seg_id) {
    int ret;

    auto src_seg_path = begin < 0 ? local_segment_path(_context.tablet_path,
                                                       _context.rowset_id.to_string(), seg_id)
                                  : BetaRowset::local_segment_path_segcompacted(
                                            _context.tablet_path, _context.rowset_id, begin, end);
    auto src_index_path_prefix = InvertedIndexDescriptor::get_index_file_path_prefix(src_seg_path);
    auto dst_seg_path = local_segment_path(_context.tablet_path, _context.rowset_id.to_string(),
                                           _num_segcompacted);
    auto dst_index_path_prefix = InvertedIndexDescriptor::get_index_file_path_prefix(dst_seg_path);

    if (_context.tablet_schema->get_inverted_index_storage_format() >=
        InvertedIndexStorageFormatPB::V2) {
        if (_context.tablet_schema->has_inverted_index() ||
            _context.tablet_schema->has_ann_index()) {
            auto src_idx_path =
                    InvertedIndexDescriptor::get_index_file_path_v2(src_index_path_prefix);
            auto dst_idx_path =
                    InvertedIndexDescriptor::get_index_file_path_v2(dst_index_path_prefix);

            ret = rename(src_idx_path.c_str(), dst_idx_path.c_str());
            if (ret) {
                return Status::Error<ROWSET_RENAME_FILE_FAILED>(
                        "failed to rename {} to {}. ret:{}, errno:{}", src_idx_path, dst_idx_path,
                        ret, errno);
            }
        }
    }
    // rename remaining inverted index files
    for (auto column : _context.tablet_schema->columns()) {
        auto index_infos = _context.tablet_schema->inverted_indexs(*column);
        for (const auto& index_info : index_infos) {
            auto index_id = index_info->index_id();
            if (_context.tablet_schema->get_inverted_index_storage_format() ==
                InvertedIndexStorageFormatPB::V1) {
                auto src_idx_path = InvertedIndexDescriptor::get_index_file_path_v1(
                        src_index_path_prefix, index_id, index_info->get_index_suffix());
                auto dst_idx_path = InvertedIndexDescriptor::get_index_file_path_v1(
                        dst_index_path_prefix, index_id, index_info->get_index_suffix());
                VLOG_DEBUG << "segcompaction skip this index. rename " << src_idx_path << " to "
                           << dst_idx_path;
                ret = rename(src_idx_path.c_str(), dst_idx_path.c_str());
                if (ret) {
                    return Status::Error<INVERTED_INDEX_RENAME_FILE_FAILED>(
                            "failed to rename {} to {}. ret:{}, errno:{}", src_idx_path,
                            dst_idx_path, ret, errno);
                }
            }
            // Erase the origin index file cache
            auto src_idx_cache_key = InvertedIndexDescriptor::get_index_file_cache_key(
                    src_index_path_prefix, index_id, index_info->get_index_suffix());
            auto dst_idx_cache_key = InvertedIndexDescriptor::get_index_file_cache_key(
                    dst_index_path_prefix, index_id, index_info->get_index_suffix());
            RETURN_IF_ERROR(InvertedIndexSearcherCache::instance()->erase(src_idx_cache_key));
            RETURN_IF_ERROR(InvertedIndexSearcherCache::instance()->erase(dst_idx_cache_key));
        }
    }
    return Status::OK();
}

Status BetaRowsetWriter::_segcompaction_if_necessary() {
    Status status = Status::OK();
    // if not doing segcompaction, just check segment number
    if (!config::enable_segcompaction || !_context.enable_segcompaction ||
        _context.tablet_schema->num_variant_columns() > 0) {
        return _check_segment_number_limit(_num_segment);
    }
    // leave _is_doing_segcompaction as the last condition
    // otherwise _segcompacting_cond will never get notified
    if (_is_doing_segcompaction.exchange(true)) {
        return status;
    }
    if (_segcompaction_status.load() != OK) {
        status = Status::Error<SEGCOMPACTION_FAILED>(
                "BetaRowsetWriter::_segcompaction_if_necessary meet invalid state, error code: {}",
                _segcompaction_status.load());
    } else {
        status = _check_segment_number_limit(_num_segcompacted);
    }
    if (status.ok() && (_num_segment - _segcompacted_point) >= config::segcompaction_batch_size) {
        SegCompactionCandidatesSharedPtr segments;
        status = _find_longest_consecutive_small_segment(segments);
        if (LIKELY(status.ok()) && (!segments->empty())) {
            LOG(INFO) << "submit segcompaction task, tablet_id:" << _context.tablet_id
                      << " rowset_id:" << _context.rowset_id << " segment num:" << _num_segment
                      << ", segcompacted_point:" << _segcompacted_point;
            status = _engine.submit_seg_compaction_task(_segcompaction_worker, segments);
            if (status.ok()) {
                return status;
            }
        }
    }
    {
        std::lock_guard lk(_is_doing_segcompaction_lock);
        _is_doing_segcompaction = false;
        _segcompacting_cond.notify_all();
    }
    return status;
}

Status BetaRowsetWriter::_segcompaction_rename_last_segments() {
    DCHECK_EQ(_is_doing_segcompaction, false);
    if (!config::enable_segcompaction) {
        return Status::OK();
    }
    if (_segcompaction_status.load() != OK) {
        return Status::Error<SEGCOMPACTION_FAILED>(
                "BetaRowsetWriter::_segcompaction_rename_last_segments meet invalid state, error "
                "code: {}",
                _segcompaction_status.load());
    }
    if (!is_segcompacted() || _segcompacted_point == _num_segment) {
        // no need if never segcompact before or all segcompacted
        return Status::OK();
    }
    // currently we only rename remaining segments to reduce wait time
    // so that transaction can be committed ASAP
    VLOG_DEBUG << "segcompaction last few segments";
    for (int32_t segid = _segcompacted_point; segid < _num_segment; segid++) {
        auto dst_segid = _num_segcompacted.load();
        RETURN_IF_ERROR(_rename_compacted_segment_plain(_segcompacted_point++));
        if (_segcompaction_worker->need_convert_delete_bitmap()) {
            _segcompaction_worker->convert_segment_delete_bitmap(
                    _context.mow_context->delete_bitmap, segid, dst_segid);
        }
    }
    return Status::OK();
}

Status BaseBetaRowsetWriter::add_rowset(RowsetSharedPtr rowset) {
    assert(rowset->rowset_meta()->rowset_type() == BETA_ROWSET);
    RETURN_IF_ERROR(rowset->link_files_to(_context.tablet_path, _context.rowset_id));
    _num_rows_written += rowset->num_rows();
    const auto& rowset_meta = rowset->rowset_meta();
    auto index_size = rowset_meta->index_disk_size();
    auto total_size = rowset_meta->total_disk_size();
    auto data_size = rowset_meta->data_disk_size();
    // corrupted index size caused by bug before 2.1.5 or 3.0.0 version
    // try to get real index size from disk.
    if (index_size < 0 || index_size > total_size * 2) {
        LOG(ERROR) << "invalid index size:" << index_size << " total size:" << total_size
                   << " data size:" << data_size << " tablet:" << rowset_meta->tablet_id()
                   << " rowset:" << rowset_meta->rowset_id();
        index_size = 0;
        auto st = rowset->get_inverted_index_size(&index_size);
        if (!st.ok()) {
            if (!st.is<NOT_FOUND>()) {
                LOG(ERROR) << "failed to get inverted index size. res=" << st;
                return st;
            }
        }
    }
    _total_data_size += data_size;
    _total_index_size += index_size;
    _num_segment += cast_set<int32_t>(rowset->num_segments());
    // append key_bounds to current rowset
    RETURN_IF_ERROR(rowset->get_segments_key_bounds(&_segments_encoded_key_bounds));
    rowset->get_num_segment_rows(&_segment_num_rows);
    _segments_key_bounds_truncated = rowset->rowset_meta()->is_segments_key_bounds_truncated();

    // TODO update zonemap
    if (rowset->rowset_meta()->has_delete_predicate()) {
        _rowset_meta->set_delete_predicate(rowset->rowset_meta()->delete_predicate());
    }
    // Update the tablet schema in the rowset metadata if the tablet schema contains a variant.
    // During the build process, _context.tablet_schema will be used as the rowset schema.
    // This situation may arise in the event of a linked schema change. If this schema is not set,
    // the subcolumns of the variant will be lost.
    if (_context.tablet_schema->num_variant_columns() > 0 && rowset->tablet_schema() != nullptr) {
        _context.tablet_schema = rowset->tablet_schema();
    }
    return Status::OK();
}

Status BaseBetaRowsetWriter::add_rowset_for_linked_schema_change(RowsetSharedPtr rowset) {
    // TODO use schema_mapping to transfer zonemap
    return add_rowset(rowset);
}

Status BaseBetaRowsetWriter::flush() {
    return _segment_creator.flush();
}

Status BaseBetaRowsetWriter::flush_memtable(Block* block, int32_t segment_id, int64_t* flush_size) {
    if (block->rows() == 0) {
        return Status::OK();
    }

    {
        SCOPED_RAW_TIMER(&_segment_writer_ns);
        RETURN_IF_ERROR(_segment_creator.flush_single_block(block, segment_id, flush_size));
    }
    return Status::OK();
}

Status BaseBetaRowsetWriter::flush_single_block(const Block* block) {
    return _segment_creator.flush_single_block(block);
}

Status BetaRowsetWriter::_wait_flying_segcompaction() {
    std::unique_lock<std::mutex> l(_is_doing_segcompaction_lock);
    uint64_t begin_wait = GetCurrentTimeMicros();
    while (_is_doing_segcompaction) {
        // change sync wait to async?
        _segcompacting_cond.wait(l);
    }
    uint64_t elapsed = GetCurrentTimeMicros() - begin_wait;
    if (elapsed >= MICROS_PER_SEC) {
        LOG(INFO) << "wait flying segcompaction finish time:" << elapsed << "us";
    }
    if (_segcompaction_status.load() != OK) {
        return Status::Error<SEGCOMPACTION_FAILED>(
                "BetaRowsetWriter meet invalid state, error code: {}",
                _segcompaction_status.load());
    }
    return Status::OK();
}

Status BetaRowsetWriter::_retry_pending_segcompaction_after_delete_bitmap() {
    while ((_num_segment - _segcompacted_point) >= config::segcompaction_batch_size) {
        RETURN_NOT_OK_STATUS_WITH_WARN(_wait_flying_segcompaction(),
                                       "segcompaction failed when build new rowset");
        const int32_t old_segcompacted_point = _segcompacted_point.load();
        const int32_t old_num_segcompacted = _num_segcompacted.load();
        RETURN_NOT_OK_STATUS_WITH_WARN(_segcompaction_if_necessary(),
                                       "segcompaction failed when build new rowset");
        RETURN_NOT_OK_STATUS_WITH_WARN(_wait_flying_segcompaction(),
                                       "segcompaction failed when build new rowset");
        if (_segcompacted_point == old_segcompacted_point &&
            _num_segcompacted == old_num_segcompacted) {
            break;
        }
    }
    return Status::OK();
}

Status BetaRowsetWriter::_finish_flying_segcompaction(bool need_final_segcompaction_retry) {
    if (need_final_segcompaction_retry) {
        return _retry_pending_segcompaction_after_delete_bitmap();
    }
    if (_segcompaction_worker->cancel()) {
        std::lock_guard lk(_is_doing_segcompaction_lock);
        _is_doing_segcompaction = false;
        _segcompacting_cond.notify_all();
        return Status::OK();
    }
    RETURN_NOT_OK_STATUS_WITH_WARN(_wait_flying_segcompaction(),
                                   "segcompaction failed when build new rowset");
    return Status::OK();
}

RowsetSharedPtr BaseBetaRowsetWriter::manual_build(const RowsetMetaSharedPtr& spec_rowset_meta) {
    if (_rowset_meta->newest_write_timestamp() == -1) {
        _rowset_meta->set_newest_write_timestamp(UnixSeconds());
    }

    build_rowset_meta_with_spec_field(*_rowset_meta, *spec_rowset_meta);
    RowsetSharedPtr rowset;
    auto status = RowsetFactory::create_rowset(_context.tablet_schema, _context.tablet_path,
                                               _rowset_meta, &rowset);
    if (!status.ok()) {
        LOG(WARNING) << "rowset init failed when build new rowset, res=" << status;
        return nullptr;
    }
    _already_built = true;
    return rowset;
}

Status BaseBetaRowsetWriter::_close_file_writers() {
    // Flush and close segment files
    RETURN_NOT_OK_STATUS_WITH_WARN(_segment_creator.close(),
                                   "failed to close segment creator when build new rowset");
    return Status::OK();
}

Status BetaRowsetWriter::_close_file_writers() {
    // Flush first to submit delete-bitmap tasks for the last segment, then wait before closing
    // the shared file writers so build() cannot race with delete-bitmap workers on close().
    RETURN_NOT_OK_STATUS_WITH_WARN(_segment_creator.flush(),
                                   "failed to flush segment creator when build new rowset");
    const bool need_final_segcompaction_retry =
            _calc_delete_bitmap_token != nullptr && config::enable_segcompaction &&
            _context.enable_segcompaction && _context.tablet_schema->num_variant_columns() == 0;
    if (_calc_delete_bitmap_token != nullptr) {
        RETURN_IF_ERROR(_calc_delete_bitmap_token->wait());
    }
    RETURN_NOT_OK_STATUS_WITH_WARN(_segment_creator.close(),
                                   "failed to close segment creator when build new rowset");
    // if _segment_start_id is not zero, that means it's a transient rowset writer for
    // MoW partial update, don't need to do segment compaction.
    if (_segment_start_id == 0) {
        RETURN_IF_ERROR(_finish_flying_segcompaction(need_final_segcompaction_retry));
        RETURN_NOT_OK_STATUS_WITH_WARN(_segcompaction_rename_last_segments(),
                                       "rename last segments failed when build new rowset");
        // segcompaction worker would do file wrier's close function in compact_segments
        if (auto& seg_comp_file_writer = _segcompaction_worker->get_file_writer();
            nullptr != seg_comp_file_writer &&
            seg_comp_file_writer->state() != io::FileWriter::State::CLOSED) {
            RETURN_NOT_OK_STATUS_WITH_WARN(seg_comp_file_writer->close(),
                                           "close segment compaction worker failed");
        }
        // process delete bitmap for mow table
        if (is_segcompacted() && _segcompaction_worker->need_convert_delete_bitmap()) {
            auto converted_delete_bitmap = _segcompaction_worker->get_converted_delete_bitmap();
            // which means the segment compaction is triggerd
            if (converted_delete_bitmap != nullptr) {
                RowsetIdUnorderedSet rowsetids;
                rowsetids.insert(rowset_id());
                context().tablet->add_sentinel_mark_to_delete_bitmap(converted_delete_bitmap.get(),
                                                                     rowsetids);
                context().mow_context->delete_bitmap->remove({rowset_id(), 0, 0},
                                                             {rowset_id(), UINT32_MAX, INT64_MAX});
                context().mow_context->delete_bitmap->merge(*converted_delete_bitmap);
            }
        }
    }
    return Status::OK();
}

Status BetaRowsetWriter::build(RowsetSharedPtr& rowset) {
    RETURN_IF_ERROR(_close_file_writers());
    RETURN_NOT_OK_STATUS_WITH_WARN(_check_segment_number_limit(_num_seg()),
                                   "too many segments when build new rowset");
    RETURN_IF_ERROR(_build_rowset_meta(_rowset_meta.get(), true));
    if (_is_pending) {
        _rowset_meta->set_rowset_state(COMMITTED);
    } else {
        _rowset_meta->set_rowset_state(VISIBLE);
    }

    if (_rowset_meta->newest_write_timestamp() == -1) {
        _rowset_meta->set_newest_write_timestamp(UnixSeconds());
    }

    _rowset_meta->set_tablet_schema(_context.tablet_schema);

    // If segment compaction occurs, the idx file info will become inaccurate.
    if ((_context.tablet_schema->has_inverted_index() || _context.tablet_schema->has_ann_index()) &&
        _num_segcompacted == 0) {
        if (auto idx_files_info = _idx_files.inverted_index_file_info(_segment_start_id);
            !idx_files_info.has_value()) [[unlikely]] {
            LOG(ERROR) << "expected inverted index files info, but none presents: "
                       << idx_files_info.error();
        } else {
            _rowset_meta->add_inverted_index_files_info(idx_files_info.value());
        }
    }

    RETURN_NOT_OK_STATUS_WITH_WARN(
            RowsetFactory::create_rowset(_context.tablet_schema, _context.tablet_path, _rowset_meta,
                                         &rowset),
            "rowset init failed when build new rowset");
    _already_built = true;
    return Status::OK();
}

int64_t BaseBetaRowsetWriter::_num_seg() const {
    return _num_segment;
}

int64_t BetaRowsetWriter::_num_seg() const {
    return is_segcompacted() ? _num_segcompacted : _num_segment;
}

Status BaseBetaRowsetWriter::_build_rowset_meta(RowsetMeta* rowset_meta, bool check_segment_num) {
    int64_t num_rows_written = 0;
    int64_t total_data_size = 0;
    int64_t total_index_size = 0;
    std::vector<KeyBoundsPB> segments_encoded_key_bounds;
    std::vector<uint32_t> segment_rows;
    std::optional<bool> segments_key_bounds_truncated;
    {
        std::lock_guard<std::mutex> lock(_segid_statistics_map_mutex);
        for (const auto& itr : _segid_statistics_map) {
            num_rows_written += itr.second.row_num;
            total_data_size += itr.second.data_size;
            total_index_size += itr.second.index_size;
            segments_encoded_key_bounds.push_back(itr.second.key_bounds);
            // segcompaction don't modify _segment_num_rows, so we need to get segment rows from _segid_statistics_map for load
            segment_rows.push_back(cast_set<uint32_t>(itr.second.row_num));
        }
        segments_key_bounds_truncated = _segments_key_bounds_truncated;
    }
    if (segment_rows.empty()) {
        // vertical compaction and linked schema change will not record segment statistics,
        // it will record segment rows in _segment_num_rows
        RETURN_IF_ERROR(get_segment_num_rows(&segment_rows));
    }

    for (auto& key_bound : _segments_encoded_key_bounds) {
        segments_encoded_key_bounds.push_back(key_bound);
    }
    if (segments_key_bounds_truncated.has_value()) {
        rowset_meta->set_segments_key_bounds_truncated(segments_key_bounds_truncated.value());
    }
    rowset_meta->set_num_segment_rows(segment_rows);
    // segment key bounds are empty in old version(before version 1.2.x). So we should not modify
    // the overlap property when key bounds are empty.
    // for mow table with cluster keys, the overlap is used for cluster keys,
    // the key_bounds is primary keys
    if (!segments_encoded_key_bounds.empty() &&
        !is_segment_overlapping(segments_encoded_key_bounds) &&
        _context.tablet_schema->cluster_key_uids().empty()) {
        rowset_meta->set_segments_overlap(NONOVERLAPPING);
    }

    auto segment_num = _num_seg();
    if (check_segment_num && config::check_segment_when_build_rowset_meta) {
        auto segments_encoded_key_bounds_size = segments_encoded_key_bounds.size();
        if (segments_encoded_key_bounds_size != segment_num) {
            return Status::InternalError(
                    "segments_encoded_key_bounds_size should  equal to _num_seg, "
                    "segments_encoded_key_bounds_size "
                    "is: {}, _num_seg is: {}",
                    segments_encoded_key_bounds_size, segment_num);
        }
        if (segment_rows.size() != segment_num) {
            return Status::InternalError(
                    "segment_rows size should equal to _num_seg, segment_rows size is: {}, "
                    "_num_seg is {}, tablet={}, rowset={}, txn={}",
                    segment_rows.size(), segment_num, _context.tablet_id,
                    _context.rowset_id.to_string(), _context.txn_id);
        }
    }

    rowset_meta->set_num_segments(segment_num);
    rowset_meta->set_num_rows(num_rows_written + _num_rows_written);
    rowset_meta->set_total_disk_size(total_data_size + _total_data_size + total_index_size +
                                     _total_index_size);
    rowset_meta->set_data_disk_size(total_data_size + _total_data_size);
    rowset_meta->set_index_disk_size(total_index_size + _total_index_size);
    bool aggregate_key_bounds = config::enable_aggregate_non_mow_key_bounds &&
                                !_context.enable_unique_key_merge_on_write;
    rowset_meta->set_segments_key_bounds(segments_encoded_key_bounds, aggregate_key_bounds);
    // TODO write zonemap to meta
    rowset_meta->set_empty((num_rows_written + _num_rows_written) == 0);
    rowset_meta->set_creation_time(time(nullptr));
    return Status::OK();
}

Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) {
    Status status;
    std::shared_ptr<RowsetMeta> tmp_rs_meta = std::make_shared<RowsetMeta>();
    tmp_rs_meta->init(_rowset_meta.get());

    status = _build_rowset_meta(tmp_rs_meta.get());
    if (!status.ok()) {
        LOG(WARNING) << "failed to build rowset meta, res=" << status;
        return status;
    }

    status = RowsetFactory::create_rowset(_context.tablet_schema, _context.tablet_path, tmp_rs_meta,
                                          &rowset_ptr);
    DBUG_EXECUTE_IF("BaseBetaRowsetWriter::_build_tmp.create_rowset_failed",
                    { status = Status::InternalError("create rowset failed"); });
    if (!status.ok()) {
        LOG(WARNING) << "rowset init failed when build new rowset, res=" << status;
        return status;
    }
    return Status::OK();
}

Status BaseBetaRowsetWriter::_create_file_writer(const std::string& path,
                                                 io::FileWriterPtr& file_writer,
                                                 bool is_index_file) {
    io::FileWriterOptions opts = _context.get_file_writer_options(is_index_file);
    Status st = _context.fs()->create_file(path, &file_writer, &opts);
    if (!st.ok()) {
        LOG(WARNING) << "failed to create writable file. path=" << path << ", err: " << st;
        return st;
    }

    DCHECK(file_writer != nullptr);
    return Status::OK();
}

Status BaseBetaRowsetWriter::create_file_writer(uint32_t segment_id, io::FileWriterPtr& file_writer,
                                                FileType file_type) {
    auto segment_path = _context.segment_path(segment_id);
    if (file_type == FileType::INVERTED_INDEX_FILE) {
        std::string prefix =
                std::string {InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)};
        std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix);
        return _create_file_writer(index_path, file_writer, true /* is_index_file */);
    } else if (file_type == FileType::SEGMENT_FILE) {
        return _create_file_writer(segment_path, file_writer, false /* is_index_file */);
    }
    return Status::Error<ErrorCode::INTERNAL_ERROR>(
            fmt::format("failed to create file = {}, file type = {}", segment_path, file_type));
}

Status BaseBetaRowsetWriter::create_index_file_writer(uint32_t segment_id,
                                                      IndexFileWriterPtr* index_file_writer) {
    RETURN_IF_ERROR(RowsetWriter::create_index_file_writer(segment_id, index_file_writer));
    // used for inverted index format v1
    (*index_file_writer)
            ->set_file_writer_opts(_context.get_file_writer_options(true /* is_index_file */));
    return Status::OK();
}

Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
        std::unique_ptr<segment_v2::SegmentWriter>* writer, int64_t begin, int64_t end) {
    DCHECK(begin >= 0 && end >= 0);
    std::string path = BetaRowset::local_segment_path_segcompacted(_context.tablet_path,
                                                                   _context.rowset_id, begin, end);
    io::FileWriterPtr file_writer;
    RETURN_IF_ERROR(_create_file_writer(path, file_writer, false /* is_index_file */));

    IndexFileWriterPtr index_file_writer;
    if (_context.tablet_schema->has_inverted_index() || _context.tablet_schema->has_ann_index()) {
        io::FileWriterPtr idx_file_writer;
        std::string prefix(InvertedIndexDescriptor::get_index_file_path_prefix(path));
        if (_context.tablet_schema->get_inverted_index_storage_format() !=
            InvertedIndexStorageFormatPB::V1) {
            std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix);
            RETURN_IF_ERROR(
                    _create_file_writer(index_path, idx_file_writer, true /* is_index_file */));
        }
        index_file_writer = std::make_unique<IndexFileWriter>(
                _context.fs(), prefix, _context.rowset_id.to_string(), _num_segcompacted,
                _context.tablet_schema->get_inverted_index_storage_format(),
                std::move(idx_file_writer), true /* can_use_ram_dir */, _context.tablet_id);
    }

    segment_v2::SegmentWriterOptions writer_options;
    writer_options.enable_unique_key_merge_on_write = _context.enable_unique_key_merge_on_write;
    writer_options.rowset_ctx = &_context;
    writer_options.write_type = _context.write_type;
    writer_options.write_type = DataWriteType::TYPE_COMPACTION;
    writer_options.max_rows_per_segment = _context.max_rows_per_segment;
    writer_options.mow_ctx = _context.mow_context;

    *writer = std::make_unique<segment_v2::SegmentWriter>(
            file_writer.get(), _num_segcompacted, _context.tablet_schema, _context.tablet,
            _context.data_dir, writer_options, index_file_writer.get());
    if (auto& seg_writer = _segcompaction_worker->get_file_writer();
        seg_writer != nullptr && seg_writer->state() != io::FileWriter::State::CLOSED) {
        RETURN_IF_ERROR(_segcompaction_worker->get_file_writer()->close());
    }
    _segcompaction_worker->get_file_writer().reset(file_writer.release());
    if (auto& idx_file_writer = _segcompaction_worker->get_inverted_index_file_writer();
        idx_file_writer != nullptr) {
        RETURN_IF_ERROR(idx_file_writer->begin_close());
        RETURN_IF_ERROR(idx_file_writer->finish_close());
    }
    _segcompaction_worker->get_inverted_index_file_writer().reset(index_file_writer.release());
    return Status::OK();
}

Status BaseBetaRowsetWriter::_check_segment_number_limit(size_t segnum) {
    DBUG_EXECUTE_IF("BetaRowsetWriter._check_segment_number_limit_too_many_segments",
                    { segnum = dp->param("segnum", 1024); });
    if (UNLIKELY(segnum > cast_set<size_t>(config::max_segment_num_per_rowset))) {
        return Status::Error<TOO_MANY_SEGMENTS>(
                "too many segments in rowset. tablet_id:{}, rowset_id:{}, max:{}, "
                "_num_segment:{}, rowset_num_rows:{}. Please check if the bucket number is too "
                "small or if the data is skewed.",
                _context.tablet_id, _context.rowset_id.to_string(),
                config::max_segment_num_per_rowset, _num_segment, get_rowset_num_rows());
    }
    return Status::OK();
}

Status BetaRowsetWriter::_check_segment_number_limit(size_t segnum) {
    DBUG_EXECUTE_IF("BetaRowsetWriter._check_segment_number_limit_too_many_segments",
                    { segnum = dp->param("segnum", 1024); });
    if (UNLIKELY(segnum > cast_set<size_t>(config::max_segment_num_per_rowset))) {
        return Status::Error<TOO_MANY_SEGMENTS>(
                "too many segments in rowset. tablet_id:{}, rowset_id:{}, max:{}, _num_segment:{}, "
                "_segcompacted_point:{}, _num_segcompacted:{}, rowset_num_rows:{}. Please check if "
                "the bucket number is too small or if the data is skewed.",
                _context.tablet_id, _context.rowset_id.to_string(),
                config::max_segment_num_per_rowset, _num_segment, _segcompacted_point,
                _num_segcompacted, get_rowset_num_rows());
    }
    return Status::OK();
}

Status BaseBetaRowsetWriter::add_segment(uint32_t segment_id, const SegmentStatistics& segstat) {
    uint32_t segid_offset = segment_id - _segment_start_id;
    bool key_bounds_truncated = false;
    SegmentStatistics stored_segstat =
            copy_segment_statistics_with_truncated_key_bounds(segstat, key_bounds_truncated);
    {
        std::lock_guard<std::mutex> lock(_segid_statistics_map_mutex);
        CHECK_EQ(_segid_statistics_map.find(segment_id) == _segid_statistics_map.end(), true);
        _segid_statistics_map.emplace(segment_id, std::move(stored_segstat));
        if (segment_id >= _segment_num_rows.size()) {
            _segment_num_rows.resize(segment_id + 1);
        }
        _segment_num_rows[segid_offset] = cast_set<uint32_t>(segstat.row_num);
        if (key_bounds_truncated) {
            _segments_key_bounds_truncated = true;
        }
    }
    VLOG_DEBUG << "_segid_statistics_map add new record. segment_id:" << segment_id
               << " row_num:" << segstat.row_num << " data_size:" << segstat.data_size
               << " index_size:" << segstat.index_size;

    {
        std::lock_guard<std::mutex> lock(_segment_set_mutex);
        _segment_set.add(segid_offset);
        while (_segment_set.contains(_num_segment)) {
            _num_segment++;
        }
    }

    if (_context.mow_context != nullptr) {
        RETURN_IF_ERROR(_generate_delete_bitmap(segment_id));
    }
    return Status::OK();
}

Status BetaRowsetWriter::add_segment(uint32_t segment_id, const SegmentStatistics& segstat) {
    RETURN_IF_ERROR(BaseBetaRowsetWriter::add_segment(segment_id, segstat));
    DBUG_EXECUTE_IF("BetaRowsetWriter.add_segment.sleep_before_segcompaction", {
        auto target_segment_id = dp->param<int64_t>("segment_id", -1);
        if (target_segment_id < 0 || segment_id == target_segment_id) {
            auto sleep_ms = dp->param<int64_t>("sleep_ms", 200);
            std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
        }
    });
    return _segcompaction_if_necessary();
}

Status BetaRowsetWriter::flush_segment_writer_for_segcompaction(
        std::unique_ptr<segment_v2::SegmentWriter>*
                writer, // NOLINT(readability-non-const-parameter): resets the caller-owned writer after finalizing it.
        uint64_t index_size, KeyBoundsPB& key_bounds) {
    uint32_t segid = (*writer)->get_segment_id();
    uint32_t row_num = (*writer)->row_count();
    uint64_t segment_size;

    auto s = (*writer)->finalize_footer(&segment_size);
    if (!s.ok()) {
        return Status::Error<WRITER_DATA_WRITE_ERROR>("failed to finalize segment: {}",
                                                      s.to_string());
    }
    int64_t inverted_index_file_size = 0;
    RETURN_IF_ERROR((*writer)->close_inverted_index(&inverted_index_file_size));

    SegmentStatistics segstat;
    segstat.row_num = row_num;
    segstat.data_size = segment_size;
    segstat.index_size = inverted_index_file_size;
    bool key_bounds_truncated = copy_key_bounds_with_truncation(key_bounds, &segstat.key_bounds);
    {
        std::lock_guard<std::mutex> lock(_segid_statistics_map_mutex);
        CHECK_EQ(_segid_statistics_map.find(segid) == _segid_statistics_map.end(), true);
        _segid_statistics_map.emplace(segid, std::move(segstat));
        if (key_bounds_truncated) {
            _segments_key_bounds_truncated = true;
        }
    }
    VLOG_DEBUG << "_segid_statistics_map add new record. segid:" << segid << " row_num:" << row_num
               << " data_size:" << PrettyPrinter::print_bytes(segment_size)
               << " index_size:" << PrettyPrinter::print_bytes(inverted_index_file_size);

    writer->reset();

    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-06-24 12:50