Coverage Report

Created: 2026-03-14 06:50

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exec/sink/writer/vwal_writer.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "exec/sink/writer/vwal_writer.h"
19
20
#include <gen_cpp/FrontendService.h>
21
#include <gen_cpp/data.pb.h>
22
23
#include <sstream>
24
25
#include "io/fs/encrypted_fs_factory.h"
26
#include "util/debug_points.h"
27
28
namespace doris {
29
30
VWalWriter::VWalWriter(int64_t db_id, int64_t tb_id, int64_t wal_id,
31
                       const std::string& import_label, WalManager* wal_manager,
32
                       std::vector<TSlotDescriptor>& slot_desc, int be_exe_version)
33
0
        : _db_id(db_id),
34
0
          _tb_id(tb_id),
35
0
          _wal_id(wal_id),
36
0
          _label(import_label),
37
0
          _wal_manager(wal_manager),
38
0
          _slot_descs(slot_desc),
39
0
          _be_exe_version(be_exe_version) {}
40
41
0
VWalWriter::~VWalWriter() {}
42
43
0
Status VWalWriter::init() {
44
0
    io::FileSystemSPtr wal_fs = io::global_local_filesystem();
45
#ifndef BE_TEST
46
    if (config::group_commit_wait_replay_wal_finish) {
47
        std::shared_ptr<std::mutex> lock = std::make_shared<std::mutex>();
48
        std::shared_ptr<std::condition_variable> cv = std::make_shared<std::condition_variable>();
49
        auto add_st = _wal_manager->add_wal_cv_map(_wal_id, lock, cv);
50
        if (!add_st.ok()) {
51
            LOG(WARNING) << "fail to add wal_id " << _wal_id << " to wal_cv_map";
52
        }
53
    }
54
    RETURN_IF_ERROR(determine_wal_fs(_db_id, _tb_id, wal_fs));
55
#endif
56
0
    RETURN_IF_ERROR(_create_wal_writer(_wal_id, wal_fs, _wal_writer));
57
0
    _wal_manager->add_wal_queue(_tb_id, _wal_id);
58
0
    std::stringstream ss;
59
0
    for (auto slot_desc : _slot_descs) {
60
0
        if (slot_desc.col_unique_id < 0) {
61
0
            continue;
62
0
        }
63
0
        ss << std::to_string(slot_desc.col_unique_id) << ",";
64
0
    }
65
0
    std::string col_ids = ss.str().substr(0, ss.str().size() - 1);
66
0
    RETURN_IF_ERROR(_wal_writer->append_header(col_ids));
67
0
    return Status::OK();
68
0
}
69
70
0
Status VWalWriter::write_wal(Block* block) {
71
0
    DBUG_EXECUTE_IF("VWalWriter.write_wal.fail",
72
0
                    { return Status::InternalError("Failed to write wal!"); });
73
0
    PBlock pblock;
74
0
    size_t uncompressed_bytes = 0, compressed_bytes = 0;
75
0
    int64_t compressed_time = 0;
76
0
    RETURN_IF_ERROR(block->serialize(_be_exe_version, &pblock, &uncompressed_bytes,
77
0
                                     &compressed_bytes, &compressed_time,
78
0
                                     segment_v2::CompressionTypePB::NO_COMPRESSION));
79
0
    RETURN_IF_ERROR(_wal_writer->append_blocks(std::vector<PBlock*> {&pblock}));
80
0
    return Status::OK();
81
0
}
82
83
0
Status VWalWriter::close() {
84
0
    if (config::group_commit_wait_replay_wal_finish) {
85
0
        std::string wal_path;
86
0
        RETURN_IF_ERROR(_wal_manager->get_wal_path(_wal_id, wal_path));
87
0
        LOG(INFO) << "close file " << wal_path;
88
0
        RETURN_IF_ERROR(_wal_manager->add_recover_wal(_db_id, _tb_id, _wal_id, wal_path));
89
0
        RETURN_IF_ERROR(_wal_manager->wait_replay_wal_finish(_wal_id));
90
0
    }
91
0
    if (_wal_writer != nullptr) {
92
0
        RETURN_IF_ERROR(_wal_writer->finalize());
93
0
    }
94
0
    return Status::OK();
95
0
}
96
97
Status VWalWriter::_create_wal_writer(int64_t wal_id, const io::FileSystemSPtr& fs,
98
0
                                      std::shared_ptr<WalWriter>& wal_writer) {
99
0
    std::string wal_path;
100
0
    RETURN_IF_ERROR(_wal_manager->get_wal_path(wal_id, wal_path));
101
0
    wal_writer = std::make_shared<WalWriter>(wal_path);
102
0
    RETURN_IF_ERROR(wal_writer->init(fs));
103
0
    return Status::OK();
104
0
}
105
} // namespace doris