Coverage Report

Created: 2026-09-14 20:34

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/rowset/rowset_fwd.h"
37
#include "storage/tablet/tablet_fwd.h"    // BaseTabletSPtr
38
#include "storage/tablet/tablet_schema.h" // TabletSchemaSPtr
39
#include "storage/utils.h"                // BINLOG_*_COL
40
41
namespace doris {
42
43
class AutoIncIDBuffer;
44
struct PartialUpdateInfo;
45
struct MowContext;
46
47
// Row binlog op type.
48
// NOTE: The value is persisted into row binlog data, so keep it stable.
49
static constexpr int64_t ROW_BINLOG_APPEND = 0;
50
static constexpr int64_t ROW_BINLOG_UPDATE = 1;
51
static constexpr int64_t ROW_BINLOG_DELETE = 2;
52
53
constexpr std::string_view kBinlogPrefix = "binlog_";
54
constexpr std::string_view kBinlogMetaPrefix = "binlog_meta_";
55
constexpr std::string_view kBinlogDataPrefix = "binlog_data_";
56
constexpr std::string_view kRowBinlogPrefix = "binlog_row_";
57
58
namespace binlog {
59
60
465
inline std::string build_before_column_name(std::string_view name) {
61
465
    std::string before_name = "__BEFORE__";
62
465
    before_name.append(name.data(), name.size());
63
465
    before_name.append("__");
64
465
    return before_name;
65
465
}
66
67
} // namespace binlog
68
69
constexpr int64_t kBinlogLsnAutoIncId = -1;
70
// used in file directory
71
constexpr std::string_view FDRowBinlogSuffix = "_row_binlog";
72
73
inline auto make_binlog_meta_key(const std::string_view tablet, int64_t version,
74
2
                                 const std::string_view rowset) {
75
2
    return fmt::format("{}meta_{}_{:020d}_{}", kBinlogPrefix, tablet, version, rowset);
76
2
}
77
78
inline auto make_binlog_meta_key(const std::string_view tablet, const std::string_view version_str,
79
0
                                 const std::string_view rowset) {
80
0
    // TODO(Drogon): use fmt::format not convert to version_num, only string with length prefix '0'
81
0
    int64_t version = std::atoll(version_str.data());
82
0
    return make_binlog_meta_key(tablet, version, rowset);
83
0
}
84
85
inline auto make_binlog_meta_key(const TabletUid& tablet_uid, int64_t version,
86
2
                                 const RowsetId& rowset_id) {
87
2
    return make_binlog_meta_key(tablet_uid.to_string(), version, rowset_id.to_string());
88
2
}
89
90
8
inline auto make_binlog_meta_key_prefix(const TabletUid& tablet_uid) {
91
8
    return fmt::format("{}meta_{}_", kBinlogPrefix, tablet_uid.to_string());
92
8
}
93
94
14
inline auto make_binlog_meta_key_prefix(const TabletUid& tablet_uid, int64_t version) {
95
14
    return fmt::format("{}meta_{}_{:020d}_", kBinlogPrefix, tablet_uid.to_string(), version);
96
14
}
97
98
inline auto make_binlog_data_key(const std::string_view tablet, int64_t version,
99
2
                                 const std::string_view rowset) {
100
2
    return fmt::format("{}data_{}_{:020d}_{}", kBinlogPrefix, tablet, version, rowset);
101
2
}
102
103
inline auto make_binlog_data_key(const std::string_view tablet, const std::string_view version,
104
0
                                 const std::string_view rowset) {
105
0
    return fmt::format("{}data_{}_{:0>20}_{}", kBinlogPrefix, tablet, version, rowset);
106
0
}
107
108
inline auto make_binlog_data_key(const TabletUid& tablet_uid, int64_t version,
109
2
                                 const RowsetId& rowset_id) {
110
2
    return make_binlog_data_key(tablet_uid.to_string(), version, rowset_id.to_string());
111
2
}
112
113
inline auto make_binlog_data_key(const TabletUid& tablet_uid, int64_t version,
114
0
                                 const std::string_view rowset_id) {
115
0
    return make_binlog_data_key(tablet_uid.to_string(), version, rowset_id);
116
0
}
117
118
0
inline auto make_binlog_data_key_prefix(const TabletUid& tablet_uid, int64_t version) {
119
0
    return fmt::format("{}data_{}_{:020d}_", kBinlogPrefix, tablet_uid.to_string(), version);
120
0
}
121
122
0
inline auto make_binlog_filename_key(const TabletUid& tablet_uid, const std::string_view version) {
123
0
    return fmt::format("{}meta_{}_{:0>20}_", kBinlogPrefix, tablet_uid.to_string(), version);
124
0
}
125
126
0
inline bool starts_with_binlog_meta(const std::string_view str) {
127
0
    auto prefix = kBinlogMetaPrefix;
128
0
    if (prefix.length() > str.length()) {
129
0
        return false;
130
0
    }
131
132
0
    return str.compare(0, prefix.length(), prefix) == 0;
133
0
}
134
135
1
inline std::string get_binlog_data_key_from_meta_key(const std::string_view meta_key) {
136
    // like "binlog_meta_6943f1585fe834b5-e542c2b83a21d0b7" => "binlog_data-6943f1585fe834b5-e542c2b83a21d0b7"
137
1
    return fmt::format("{}data_{}", kBinlogPrefix, meta_key.substr(kBinlogMetaPrefix.length()));
138
1
}
139
140
// Allocate per-row LSNs.
141
// The caller must provide a valid auto-inc buffer (typically from GlobalAutoIncBuffers).
142
inline Status allocate_lsn(const std::shared_ptr<AutoIncIDBuffer>& lsn_buffer, size_t num_rows,
143
210
                           std::vector<int64_t>& lsn_ids) {
144
210
    if (lsn_buffer == nullptr) {
145
0
        return Status::InternalError("try to get lsn buffer but null");
146
0
    }
147
210
    DCHECK(num_rows > 0);
148
149
210
    std::vector<std::pair<int64_t, size_t>> ranges;
150
210
    RETURN_IF_ERROR(lsn_buffer->sync_request_ids(num_rows, &ranges));
151
152
210
    lsn_ids.clear();
153
210
    lsn_ids.reserve(num_rows);
154
210
    for (const auto& [start, length] : ranges) {
155
602
        for (size_t i = 0; i < length; ++i) {
156
392
            DCHECK_LE(start, std::numeric_limits<int64_t>::max() - static_cast<int64_t>(i));
157
392
            lsn_ids.push_back(start + static_cast<int64_t>(i));
158
392
        }
159
210
    }
160
210
    DCHECK_EQ(lsn_ids.size(), num_rows);
161
210
    return Status::OK();
162
210
}
163
164
constexpr int64_t kTsoLogicalBits = 18;
165
166
2.08M
inline int64_t extract_tso_physical_time(int64_t tso) {
167
2.08M
    return tso <= 0 ? 0 : tso >> kTsoLogicalBits;
168
2.08M
}
169
170
2.08M
inline int64_t row_binlog_ttl_cutoff_tso(int64_t reference_tso, int64_t ttl_seconds) {
171
2.08M
    DCHECK_GT(reference_tso, 0);
172
2.08M
    DCHECK_GE(ttl_seconds, 0);
173
2.08M
    const int64_t reference_ms = extract_tso_physical_time(reference_tso);
174
2.08M
    if (ttl_seconds > reference_ms / 1000) {
175
0
        return 0;
176
0
    }
177
2.08M
    constexpr int64_t kMaxLogical = (1L << kTsoLogicalBits) - 1;
178
2.08M
    return ((reference_ms - ttl_seconds * 1000) << kTsoLogicalBits) | kMaxLogical;
179
2.08M
}
180
181
namespace segment_v2 {
182
183
class SegmentAllocatedLsnMap {
184
public:
185
324
    void insert_segment_allocated_lsns(int64_t seg_id, ConstAllocatedLsnVectorSharedPtr lsn_ids) {
186
324
        std::lock_guard<std::mutex> l(_mutex);
187
324
        _seg_id_to_lsn_ids.emplace(seg_id, std::move(lsn_ids));
188
324
    }
189
190
570
    void remove_segment(int64_t seg_id) {
191
570
        std::lock_guard<std::mutex> l(_mutex);
192
570
        _seg_id_to_lsn_ids.erase(seg_id);
193
570
    }
194
195
    bool contains_segment(int64_t seg_id) const {
196
        std::lock_guard<std::mutex> l(_mutex);
197
        return _seg_id_to_lsn_ids.count(seg_id) > 0;
198
    }
199
200
311
    ConstAllocatedLsnVectorSharedPtr get_segment_allocated_lsns(int64_t seg_id) const {
201
311
        std::lock_guard<std::mutex> l(_mutex);
202
311
        auto it = _seg_id_to_lsn_ids.find(seg_id);
203
18.4E
        DCHECK(it != _seg_id_to_lsn_ids.end())
204
18.4E
                << "SegmentAllocatedLsnMap::get_segment_allocated_lsns missing seg_id=" << seg_id
205
18.4E
                << ", existing_seg_ids=[" << ([&] {
206
0
                       std::string s;
207
0
                       for (const auto& [id, _] : _seg_id_to_lsn_ids) {
208
0
                           if (!s.empty()) {
209
0
                               s.push_back(',');
210
0
                           }
211
0
                           s.append(std::to_string(id));
212
0
                       }
213
0
                       return s;
214
0
                   }())
215
18.4E
                << "]";
216
311
        return it->second;
217
311
    }
218
219
private:
220
    mutable std::mutex _mutex;
221
    std::map<int64_t, ConstAllocatedLsnVectorSharedPtr> _seg_id_to_lsn_ids;
222
};
223
224
struct SegmentWriteBinlogOptions {
225
public:
226
    bool write_before = false;
227
228
    // source context, used for retrieving historical row and building binlog<row> block
229
    struct SourceWriteDataOptions {
230
        BaseTabletSPtr base_tablet = nullptr;
231
        TabletSchemaSPtr tablet_schema = nullptr;
232
        std::shared_ptr<PartialUpdateInfo> partial_update_info;
233
        std::shared_ptr<MowContext> mow_context;
234
        bool is_transient_rowset_writer = false;
235
        DataWriteType source_write_type = DataWriteType::TYPE_DEFAULT;
236
        // input rowset's row-binlog, read LSN from it at publish time
237
        RowsetSharedPtr row_binlog_rowset = nullptr;
238
    } source;
239
};
240
} // namespace segment_v2
241
242
} // namespace doris