be/src/storage/rowset/segcompaction.h

// 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 <memory>
#include <vector>

#include "common/status.h"
#include "io/fs/file_reader_writer_fwd.h"
#include "storage/index/index_file_writer.h"
#include "storage/merger.h"
#include "storage/segment/segment.h"
#include "storage/simple_rowid_conversion.h"
#include "storage/tablet/tablet.h"

namespace doris {
class Schema;

namespace segment_v2 {
class SegmentWriter;
} // namespace segment_v2
class RowSourcesBuffer;
class VerticalBlockReader;
struct OlapReaderStatistics;

using SegCompactionCandidates = std::vector<segment_v2::SegmentSharedPtr>;
using SegCompactionCandidatesSharedPtr = std::shared_ptr<SegCompactionCandidates>;

class BetaRowsetWriter;

class SegcompactionWorker {
    friend class BetaRowsetWriter;

public:
    explicit SegcompactionWorker(BetaRowsetWriter* writer);

    ~SegcompactionWorker() {
        DCHECK(_seg_compact_mem_tracker != nullptr);
        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_seg_compact_mem_tracker);
        if (_rowid_conversion) {
            _rowid_conversion.reset();
        }
    }

    void compact_segments(SegCompactionCandidatesSharedPtr segments);

    bool need_convert_delete_bitmap();

    void convert_segment_delete_bitmap(DeleteBitmapPtr src_delete_bitmap, uint32_t src_seg_id,
                                       uint32_t dest_seg_id);
    void convert_segment_delete_bitmap(DeleteBitmapPtr src_delete_bitmap, uint32_t src_begin,
                                       uint32_t src_end, uint32_t dest_seg_id);
    DeleteBitmapPtr get_converted_delete_bitmap() { return _converted_delete_bitmap; }

    io::FileWriterPtr& get_file_writer() { return _file_writer; }
    IndexFileWriterPtr& get_inverted_index_file_writer() { return _index_file_writer; }

    // set the cancel flag, tasks already started will not be cancelled.
    bool cancel();

    void init_mem_tracker(const RowsetWriterContext& rowset_writer_context);

private:
    Status _create_segment_writer_for_segcompaction(
            std::unique_ptr<segment_v2::SegmentWriter>* writer, uint32_t begin, uint32_t end);
    Status _get_segcompaction_reader(SegCompactionCandidatesSharedPtr segments,
                                     TabletSharedPtr tablet, std::shared_ptr<Schema> schema,
                                     OlapReaderStatistics* stat, RowSourcesBuffer& row_sources_buf,
                                     bool is_key, std::vector<uint32_t>& return_columns,
                                     std::vector<uint32_t>& key_group_cluster_key_idxes,
                                     std::unique_ptr<VerticalBlockReader>* reader);
    std::unique_ptr<segment_v2::SegmentWriter> _create_segcompaction_writer(uint32_t begin,
                                                                            uint32_t end);
    Status _delete_original_segments(uint32_t begin, uint32_t end);
    Status _check_correctness(OlapReaderStatistics& reader_stat, Merger::Statistics& merger_stat,
                              uint32_t begin, uint32_t end, bool is_mow_with_cluster_keys);
    Status _do_compact_segments(SegCompactionCandidatesSharedPtr segments);

private:
    //TODO(zhengyu): current impl depends heavily on the access to feilds of BetaRowsetWriter
    // Currently cloud storage engine doesn't need segcompaction
    BetaRowsetWriter* _writer = nullptr;
    io::FileWriterPtr _file_writer;
    IndexFileWriterPtr _index_file_writer = nullptr;

    // for unique key mow table
    std::unique_ptr<SimpleRowIdConversion> _rowid_conversion = nullptr;
    DeleteBitmapPtr _converted_delete_bitmap;
    std::shared_ptr<MemTrackerLimiter> _seg_compact_mem_tracker = nullptr;

    // the state is not mutable when 1)actual compaction operation started or 2) cancelled
    std::atomic<bool> _is_compacting_state_mutable = true;
};
} // namespace doris