Coverage Report

Created: 2026-06-14 05:10

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/binlog.h
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
#pragma once
19
20
#include <fmt/format.h>
21
22
#include <limits>
23
#include <map>
24
#include <memory>
25
#include <mutex>
26
#include <string>
27
#include <string_view>
28
#include <utility>
29
#include <vector>
30
31
#include "common/logging.h" // DCHECK
32
#include "common/status.h"
33
#include "exec/sink/autoinc_buffer.h"
34
#include "storage/olap_common.h"
35
#include "storage/olap_define.h"          // DataWriteType
36
#include "storage/tablet/tablet_schema.h" // TabletSchemaSPtr
37
#include "storage/utils.h"                // BINLOG_TIMESTAMP_COL
38
39
namespace doris {
40
41
class AutoIncIDBuffer;
42
struct PartialUpdateInfo;
43
struct MowContext;
44
45
// Row binlog op type.
46
// NOTE: The value is persisted into row binlog data, so keep it stable.
47
static constexpr int64_t ROW_BINLOG_APPEND = 0;
48
static constexpr int64_t ROW_BINLOG_UPDATE = 1;
49
static constexpr int64_t ROW_BINLOG_DELETE = 2;
50
51
constexpr std::string_view kBinlogPrefix = "binlog_";
52
constexpr std::string_view kBinlogMetaPrefix = "binlog_meta_";
53
constexpr std::string_view kBinlogDataPrefix = "binlog_data_";
54
constexpr std::string_view kRowBinlogPrefix = "binlog_row_";
55
// Alias to BINLOG_LSN_COL in storage/utils.h to keep a single source of truth.
56
static const std::string& kRowBinlogLsnColName = BINLOG_LSN_COL;
57
// Alias to BINLOG_TIMESTAMP_COL in storage/utils.h to keep a single source of truth.
58
static const std::string& kRowBinlogTimestampColName = BINLOG_TIMESTAMP_COL;
59
constexpr std::string_view kRowBinlogOpColName = "__DORIS_BINLOG_OP__";
60
61
constexpr int64_t kBinlogLsnAutoIncId = -1;
62
// used in file directory
63
constexpr std::string_view FDRowBinlogSuffix = "_row_binlog";
64
65
inline auto make_binlog_meta_key(const std::string_view tablet, int64_t version,
66
2
                                 const std::string_view rowset) {
67
2
    return fmt::format("{}meta_{}_{:020d}_{}", kBinlogPrefix, tablet, version, rowset);
68
2
}
69
70
inline auto make_binlog_meta_key(const std::string_view tablet, const std::string_view version_str,
71
0
                                 const std::string_view rowset) {
72
0
    // TODO(Drogon): use fmt::format not convert to version_num, only string with length prefix '0'
73
0
    int64_t version = std::atoll(version_str.data());
74
0
    return make_binlog_meta_key(tablet, version, rowset);
75
0
}
76
77
inline auto make_binlog_meta_key(const TabletUid& tablet_uid, int64_t version,
78
2
                                 const RowsetId& rowset_id) {
79
2
    return make_binlog_meta_key(tablet_uid.to_string(), version, rowset_id.to_string());
80
2
}
81
82
6
inline auto make_binlog_meta_key_prefix(const TabletUid& tablet_uid) {
83
6
    return fmt::format("{}meta_{}_", kBinlogPrefix, tablet_uid.to_string());
84
6
}
85
86
10
inline auto make_binlog_meta_key_prefix(const TabletUid& tablet_uid, int64_t version) {
87
10
    return fmt::format("{}meta_{}_{:020d}_", kBinlogPrefix, tablet_uid.to_string(), version);
88
10
}
89
90
inline auto make_binlog_data_key(const std::string_view tablet, int64_t version,
91
2
                                 const std::string_view rowset) {
92
2
    return fmt::format("{}data_{}_{:020d}_{}", kBinlogPrefix, tablet, version, rowset);
93
2
}
94
95
inline auto make_binlog_data_key(const std::string_view tablet, const std::string_view version,
96
0
                                 const std::string_view rowset) {
97
0
    return fmt::format("{}data_{}_{:0>20}_{}", kBinlogPrefix, tablet, version, rowset);
98
0
}
99
100
inline auto make_binlog_data_key(const TabletUid& tablet_uid, int64_t version,
101
2
                                 const RowsetId& rowset_id) {
102
2
    return make_binlog_data_key(tablet_uid.to_string(), version, rowset_id.to_string());
103
2
}
104
105
inline auto make_binlog_data_key(const TabletUid& tablet_uid, int64_t version,
106
0
                                 const std::string_view rowset_id) {
107
0
    return make_binlog_data_key(tablet_uid.to_string(), version, rowset_id);
108
0
}
109
110
0
inline auto make_binlog_data_key_prefix(const TabletUid& tablet_uid, int64_t version) {
111
0
    return fmt::format("{}data_{}_{:020d}_", kBinlogPrefix, tablet_uid.to_string(), version);
112
0
}
113
114
0
inline auto make_binlog_filename_key(const TabletUid& tablet_uid, const std::string_view version) {
115
0
    return fmt::format("{}meta_{}_{:0>20}_", kBinlogPrefix, tablet_uid.to_string(), version);
116
0
}
117
118
0
inline bool starts_with_binlog_meta(const std::string_view str) {
119
0
    auto prefix = kBinlogMetaPrefix;
120
0
    if (prefix.length() > str.length()) {
121
0
        return false;
122
0
    }
123
124
0
    return str.compare(0, prefix.length(), prefix) == 0;
125
0
}
126
127
1
inline std::string get_binlog_data_key_from_meta_key(const std::string_view meta_key) {
128
    // like "binlog_meta_6943f1585fe834b5-e542c2b83a21d0b7" => "binlog_data-6943f1585fe834b5-e542c2b83a21d0b7"
129
1
    return fmt::format("{}data_{}", kBinlogPrefix, meta_key.substr(kBinlogMetaPrefix.length()));
130
1
}
131
132
0
inline auto make_row_binlog_key_prefix(const TabletUid& tablet_uid, const RowsetId& rowset_id) {
133
0
    return fmt::format("{}{}_{}_", kRowBinlogPrefix, tablet_uid.to_string(), rowset_id.to_string());
134
0
}
135
136
inline auto make_row_binlog_key(const TabletUid& tablet_uid, const RowsetId& rowset_id,
137
9
                                const RowsetId& binlog_rowset_id) {
138
9
    return fmt::format("{}{}_{}_{}", kRowBinlogPrefix, tablet_uid.to_string(),
139
9
                       rowset_id.to_string(), binlog_rowset_id.to_string());
140
9
}
141
142
// Allocate per-row LSNs for row-binlog data.
143
// The caller must provide a valid auto-inc buffer (typically from GlobalAutoIncBuffers).
144
inline Status allocate_binlog_lsn(const std::shared_ptr<AutoIncIDBuffer>& lsn_buffer,
145
5
                                  size_t num_rows, std::shared_ptr<std::vector<int64_t>>* lsn_ids) {
146
5
    if (lsn_buffer == nullptr) {
147
0
        return Status::InternalError("binlog<row> try to get lsn buffer but null");
148
0
    }
149
5
    DCHECK(lsn_ids != nullptr);
150
5
    DCHECK(num_rows > 0);
151
152
5
    std::vector<std::pair<int64_t, size_t>> ranges;
153
5
    RETURN_IF_ERROR(lsn_buffer->sync_request_ids(num_rows, &ranges));
154
155
5
    auto ids = std::make_shared<std::vector<int64_t>>();
156
5
    ids->reserve(num_rows);
157
5
    for (const auto& [start, length] : ranges) {
158
12
        for (size_t i = 0; i < length; ++i) {
159
7
            DCHECK_LE(start, std::numeric_limits<int64_t>::max() - static_cast<int64_t>(i));
160
7
            ids->push_back(start + static_cast<int64_t>(i));
161
7
        }
162
5
    }
163
5
    DCHECK_EQ(ids->size(), num_rows);
164
5
    *lsn_ids = std::move(ids);
165
5
    return Status::OK();
166
5
}
167
168
constexpr int64_t kTsoLogicalBits = 18;
169
170
0
inline int64_t extract_tso_physical_time(int64_t tso) {
171
0
    return tso <= 0 ? 0 : tso >> kTsoLogicalBits;
172
0
}
173
174
0
inline int128_t make_row_binlog_lsn(int64_t tso, int128_t row_id) {
175
0
    static constexpr int128_t kLow64Mask = (static_cast<int128_t>(1) << 64) - 1;
176
0
    return (static_cast<int128_t>(tso) << 64) | (row_id & kLow64Mask);
177
0
}
178
179
namespace segment_v2 {
180
181
class SegmentWriteBinlogLsnMap {
182
public:
183
5
    void insert_seg_lsn(int64_t seg_id, std::shared_ptr<std::vector<int64_t>> lsn_ids) {
184
5
        std::lock_guard<std::mutex> l(_mutex);
185
5
        _seg_id_to_lsn_ids.emplace(seg_id, std::move(lsn_ids));
186
5
    }
187
188
2
    void remove_seg(int64_t seg_id) {
189
2
        std::lock_guard<std::mutex> l(_mutex);
190
2
        _seg_id_to_lsn_ids.erase(seg_id);
191
2
    }
192
193
3
    std::shared_ptr<const std::vector<int64_t>> get_seg_lsn(int64_t seg_id) const {
194
3
        std::lock_guard<std::mutex> l(_mutex);
195
3
        auto it = _seg_id_to_lsn_ids.find(seg_id);
196
3
        CHECK(it != _seg_id_to_lsn_ids.end())
197
0
                << "SegmentWriteBinlogLsnMap::get_seg_lsn missing seg_id=" << seg_id
198
0
                << ", existing_seg_ids=[" << ([&] {
199
0
                       std::string s;
200
0
                       for (const auto& [id, _] : _seg_id_to_lsn_ids) {
201
0
                           if (!s.empty()) {
202
0
                               s.push_back(',');
203
0
                           }
204
0
                           s.append(std::to_string(id));
205
0
                       }
206
0
                       return s;
207
0
                   }())
208
0
                << "]";
209
3
        return it->second;
210
3
    }
211
212
private:
213
    mutable std::mutex _mutex;
214
    std::map<int64_t, std::shared_ptr<std::vector<int64_t>>> _seg_id_to_lsn_ids;
215
};
216
217
struct SegmentWriteBinlogOptions {
218
public:
219
    bool write_before = false;
220
221
    // source context, used for retrieving historical row and building binlog<row> block
222
    struct SourceWriteDataOptions {
223
        TabletSchemaSPtr tablet_schema = nullptr;
224
        std::shared_ptr<PartialUpdateInfo> partial_update_info;
225
        std::shared_ptr<MowContext> mow_context;
226
        bool is_transient_rowset_writer = false;
227
        DataWriteType source_write_type = DataWriteType::TYPE_DEFAULT;
228
    } source;
229
230
5
    void insert_seg_lsn(int64_t seg_id, std::shared_ptr<std::vector<int64_t>> lsn_ids) {
231
5
        DCHECK(lsn_map != nullptr);
232
5
        lsn_map->insert_seg_lsn(seg_id, std::move(lsn_ids));
233
5
    }
234
235
2
    void remove_seg(int64_t seg_id) {
236
2
        DCHECK(lsn_map != nullptr);
237
2
        lsn_map->remove_seg(seg_id);
238
2
    }
239
240
3
    std::shared_ptr<const std::vector<int64_t>> get_seg_lsn(int64_t seg_id) const {
241
        DCHECK(lsn_map != nullptr);
242
3
        return lsn_map->get_seg_lsn(seg_id);
243
3
    }
244
245
    // Shared LSN storage for row-binlog writers.
246
    // Keep it as a pointer so SegmentWriteBinlogOptions stays copyable.
247
    std::shared_ptr<SegmentWriteBinlogLsnMap> lsn_map =
248
            std::make_shared<SegmentWriteBinlogLsnMap>();
249
};
250
} // namespace segment_v2
251
252
} // namespace doris