Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./extension/vdb_writer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <array>
28 : : #include <cstdint>
29 : : #include <cstdio>
30 : : #include <filesystem>
31 : : #include <fstream>
32 : : #include <memory>
33 : : #include <mutex>
34 : : #include <string>
35 : : #include <string_view>
36 : : #include <unordered_map>
37 : : #include <unordered_set>
38 : : #include <utility>
39 : : #include <vector>
40 : :
41 : : #include "./base/elapsed_timer.h"
42 : : #include "./base/helpers.h"
43 : : #include "./base/logger.h"
44 : : #include "./version.h"
45 : :
46 : : // json
47 : : #include <nlohmann/json.hpp>
48 : :
49 : : #ifdef VLINK_ENABLE_SQLITE
50 : : #include <sqlite3.h>
51 : : #endif
52 : :
53 : : // schema_plugin
54 : : #include "./extension/schema_plugin_interface.h"
55 : :
56 : : namespace vlink {
57 : :
58 : 6560 : [[maybe_unused]] static constexpr int get_column(int column) noexcept { return column + 1; }
59 : :
60 : : static constexpr int kSyncWriteInterval = 1000; // ms
61 : : static constexpr int kCompressMaxIgnoreCnt = 5;
62 : : static constexpr double kCompressMaxRatio = 0.95;
63 : :
64 : : // VDBWriter::Impl
65 : : struct VDBWriter::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
66 : : // UrlMsgInfo
67 : : struct UrlMsgInfo final {
68 : : int index{0};
69 : : size_t count{0};
70 : : size_t size{0};
71 : : int64_t first_timestamp{-1};
72 : : int64_t last_timestamp{-1};
73 : : int64_t previous_segment_last_timestamp{-1};
74 : : double freq{0};
75 : : double loss{0};
76 : : std::string url_type;
77 : : std::string ser_type;
78 : : SchemaType schema_type{SchemaType::kUnknown};
79 : :
80 : : bool operator<(const UrlMsgInfo& target) const noexcept { return index < target.index; }
81 : : };
82 : :
83 : : struct WriteStateSnapshot final {
84 : : int64_t current_row{0};
85 : : int64_t current_size{0};
86 : : bool has_oversize{false};
87 : : int64_t last_timestamp{0};
88 : : std::vector<std::string> total_url_list;
89 : : int64_t total_current_row{0};
90 : : int64_t total_current_size{0};
91 : : int64_t total_timestamp{0};
92 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
93 : : std::unordered_set<std::string> ser_map;
94 : : std::unordered_map<std::string, UrlMsgInfo> total_url_map;
95 : : std::unordered_map<std::string, SchemaData> total_schema_map;
96 : : std::unordered_map<std::string, int64_t> compress_ignore_map;
97 : : std::string write_url_type;
98 : :
99 : 465 : explicit WriteStateSnapshot(const VDBWriter::Impl& impl)
100 : 465 : : current_row(impl.current_row),
101 : 465 : current_size(impl.current_size),
102 : 465 : has_oversize(impl.has_oversize),
103 : 465 : last_timestamp(impl.last_timestamp),
104 : 465 : total_url_list(impl.total_url_list),
105 : 465 : total_current_row(impl.total_current_row),
106 : 465 : total_current_size(impl.total_current_size),
107 : 465 : total_timestamp(impl.total_timestamp),
108 [ + - ]: 465 : url_map(impl.url_map),
109 [ + - ]: 465 : ser_map(impl.ser_map),
110 [ + - ]: 465 : total_url_map(impl.total_url_map),
111 [ + - ]: 465 : total_schema_map(impl.total_schema_map),
112 [ + - ]: 465 : compress_ignore_map(impl.compress_ignore_map),
113 [ + - ]: 465 : write_url_type(impl.write_url_type) {}
114 : :
115 : 2 : void restore(VDBWriter::Impl& impl) const {
116 : 2 : impl.current_row = current_row;
117 : 2 : impl.current_size = current_size;
118 : 2 : impl.has_oversize = has_oversize;
119 : 2 : impl.last_timestamp = last_timestamp;
120 : 2 : impl.total_url_list = total_url_list;
121 : 2 : impl.total_current_row = total_current_row;
122 : 2 : impl.total_current_size = total_current_size;
123 : 2 : impl.total_timestamp = total_timestamp;
124 : 2 : impl.url_map = url_map;
125 : 2 : impl.ser_map = ser_map;
126 : 2 : impl.total_url_map = total_url_map;
127 : 2 : impl.total_schema_map = total_schema_map;
128 : 2 : impl.compress_ignore_map = compress_ignore_map;
129 : 2 : impl.write_url_type = write_url_type;
130 : 2 : }
131 : : };
132 : :
133 : : struct MemoryCharge final {
134 : 13 : MemoryCharge(std::atomic<int64_t>& counter, int64_t bytes) : value(&counter), size(bytes) {}
135 : :
136 : 26 : MemoryCharge(MemoryCharge&& other) noexcept : value(std::exchange(other.value, nullptr)), size(other.size) {}
137 : :
138 : 39 : ~MemoryCharge() {
139 [ + + ]: 39 : if (value) {
140 : 13 : value->fetch_sub(size, std::memory_order_relaxed);
141 : : }
142 : 39 : }
143 : :
144 : : std::atomic<int64_t>* value;
145 : : int64_t size;
146 : :
147 : : VLINK_DISALLOW_COPY_AND_ASSIGN(MemoryCharge)
148 : : };
149 : :
150 : : std::atomic_bool is_dumping{false};
151 : : std::atomic_bool is_split_mode{false};
152 : : std::atomic<int> split_index{0};
153 : : std::atomic<int64_t> memory_size{0};
154 : : std::atomic_bool in_cached{false};
155 : : std::atomic<int64_t> cached_size{0};
156 : : std::atomic_bool quit_flag{false};
157 : :
158 : : std::string path;
159 : : std::filesystem::path split_output_dir;
160 : : std::string base_dir;
161 : : std::string base_name;
162 : : BagWriter::Config config;
163 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
164 : :
165 : : int64_t current_row{0};
166 : : int64_t current_size{0};
167 : : bool has_oversize{false};
168 : :
169 : : int64_t last_timestamp{0};
170 : :
171 : : BagWriter::SystemClock time_start;
172 : : BagWriter::SystemClock time_current;
173 : : int64_t start_timestamp{0};
174 : :
175 : : std::vector<std::string> split_file_list;
176 : : bool split_before{false};
177 : : bool split_first{false};
178 : :
179 : : std::vector<std::string> total_url_list;
180 : : int64_t total_current_row{0};
181 : : int64_t total_current_size{0};
182 : : int64_t total_timestamp{0};
183 : :
184 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
185 : : std::unordered_set<std::string> ser_map;
186 : : std::unordered_map<std::string, UrlMsgInfo> total_url_map;
187 : : std::unordered_map<std::string, SchemaData> total_schema_map;
188 : :
189 : : BagWriter::SplitCallback split_callback;
190 : : BagWriter::SchemaCallback schema_callback;
191 : : std::string split_filename;
192 : : std::mutex split_mtx;
193 : :
194 : : std::string app_name;
195 : : std::string tag_name;
196 : : int32_t timezone_diff{0};
197 : :
198 : : bool enable_compressed{false};
199 : : std::mutex write_mtx;
200 : :
201 : : ElapsedTimer cache_timer;
202 : : Timer check_timer;
203 : :
204 : : ElapsedTimer sync_timer;
205 : :
206 : : std::unordered_map<std::string, int64_t> compress_ignore_map;
207 : : std::unique_ptr<WriteStateSnapshot> cache_snapshot;
208 : :
209 : : std::string write_url_type;
210 : :
211 : : // database
212 : : #ifdef VLINK_ENABLE_SQLITE
213 : : ::sqlite3* db{nullptr};
214 : : ::sqlite3_stmt* schemas_stmt{nullptr};
215 : : ::sqlite3_stmt* datas_stmt{nullptr};
216 : : ::sqlite3_stmt* urls_stmt{nullptr};
217 : : ::sqlite3_stmt* update_complete_stmt{nullptr};
218 : : ::sqlite3_stmt* update_header_stmt{nullptr};
219 : : ::sqlite3_stmt* update_url_loss_stmt{nullptr};
220 : : ::sqlite3_stmt* update_url_meta_stmt{nullptr};
221 : : ::sqlite3_stmt* update_urls_stmt{nullptr};
222 : : #endif
223 : :
224 : : // schema plugin interface
225 : : SchemaPluginInterface* schema_plugin_interface{nullptr};
226 : : };
227 : :
228 : : // VDBWriter
229 : 122 : VDBWriter::VDBWriter(const std::string& path, const Config& config)
230 [ + - ]: 122 : : BagWriter(path, config), impl_(std::make_unique<Impl>()) {
231 [ + - + - ]: 122 : set_name("VDBWriter");
232 : :
233 [ + - ]: 122 : impl_->url_map.reserve(128);
234 [ + - ]: 122 : impl_->ser_map.reserve(128);
235 [ + - + - ]: 122 : url_loss_map_ref().reserve(128);
236 [ + - ]: 122 : impl_->total_url_map.reserve(128);
237 [ + - + - ]: 122 : total_url_loss_map_ref().reserve(128);
238 [ + - ]: 122 : impl_->total_schema_map.reserve(128);
239 : :
240 [ + - ]: 122 : impl_->schema_plugin_interface = get_schema_interface();
241 : :
242 [ + - + - ]: 122 : impl_->app_name = get_default_app_name();
243 : :
244 [ + + ]: 122 : if (config.tag_name.empty()) {
245 [ + - + - ]: 24 : impl_->tag_name = get_default_tag_name();
246 : : } else {
247 [ + - ]: 98 : impl_->tag_name = config.tag_name;
248 : : }
249 : :
250 [ + - ]: 122 : impl_->timezone_diff = get_default_timezone_diff();
251 : :
252 [ + - ]: 122 : impl_->path = path;
253 [ + - ]: 122 : impl_->config = config;
254 : :
255 [ + + + + ]: 122 : impl_->enable_compressed = impl_->config.compress == kCompressAuto || impl_->config.compress == kCompressLzav;
256 : :
257 [ + + ]: 122 : if VUNLIKELY (impl_->config.max_task_depth <= 0) {
258 : 2 : impl_->config.max_task_depth = BagWriter::Config().max_task_depth;
259 : : }
260 : :
261 [ + - ]: 122 : reset_lockfree_capacity();
262 : :
263 [ + + ]: 122 : if (!config.sync_mode) {
264 [ + - ]: 14 : impl_->check_timer.attach(this);
265 [ + - ]: 14 : impl_->check_timer.set_interval(kSyncWriteInterval);
266 [ + - + - ]: 14 : impl_->check_timer.start([this]() {
267 : : // LCOV_EXCL_START GCOVR_EXCL_START
268 : : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
269 : : return;
270 : : }
271 : :
272 : : if (impl_->cache_timer.get() > 3000) {
273 : : std::lock_guard lock(impl_->write_mtx);
274 : : sync_cache();
275 : : }
276 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
277 : : });
278 : : }
279 : :
280 : : try {
281 : : #ifdef _WIN32
282 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
283 : : std::string suffix = Helpers::path_to_string(file_path.extension());
284 : : #else
285 [ + - ]: 122 : std::filesystem::path file_path(path);
286 [ + - + - ]: 122 : std::string suffix = file_path.extension().string();
287 : : #endif
288 : :
289 [ + - ]: 122 : const auto parent_path = file_path.parent_path();
290 : :
291 : 643 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
292 : :
293 [ + + ]: 122 : if (suffix == ".vdbx") {
294 [ + - + + ]: 33 : if (std::filesystem::exists(file_path)) {
295 : : try {
296 : 4 : nlohmann::json root_json;
297 : :
298 : : {
299 [ + - ]: 4 : std::ifstream filex(file_path);
300 : :
301 [ + + ]: 4 : filex >> root_json;
302 : :
303 [ + - ]: 3 : filex.close();
304 : 4 : }
305 : :
306 [ + - + - ]: 3 : nlohmann::json files_json = root_json["VLinkFiles"];
307 : :
308 [ + - + - : 12 : for (const auto& file_info : files_json) {
+ - + + ]
309 [ + + ]: 9 : if (!file_info.is_string()) {
310 : 4 : continue;
311 : : }
312 : :
313 [ + - ]: 8 : const auto stale_file_name = file_info.get<std::string>();
314 : : #ifdef _WIN32
315 : : const std::filesystem::path stale_file_path(Helpers::string_to_wstring(stale_file_name));
316 : : #else
317 [ + - ]: 8 : const std::filesystem::path stale_file_path(stale_file_name);
318 : : #endif
319 : :
320 [ + - + - : 24 : if (stale_file_path.empty() || stale_file_path == "." || stale_file_path == ".." ||
+ - + - +
- + + + -
+ + - - -
- ]
321 [ + - + - : 16 : stale_file_path != stale_file_path.filename()) {
+ - - - ]
322 [ + - + - ]: 6 : CLOG_W("VDBWriter: Ignore unsafe split file path [%s].", stale_file_name.c_str());
323 : 3 : continue;
324 : 3 : }
325 : :
326 [ + - ]: 5 : const auto stale_output_path = parent_path / stale_file_path;
327 : 5 : std::error_code remove_ec;
328 : 5 : std::filesystem::remove(stale_output_path, remove_ec);
329 : :
330 [ + + ]: 5 : if VUNLIKELY (remove_ec) {
331 [ + - + - : 2 : CLOG_W("VDBWriter: Failed to remove stale split path [%s]: %s.", stale_file_name.c_str(),
+ - ]
332 : : remove_ec.message().c_str());
333 : : }
334 [ + + + + ]: 11 : }
335 : :
336 [ + - ]: 3 : std::filesystem::remove(file_path);
337 [ - + ]: 5 : } catch (const nlohmann::json::exception& e) {
338 [ + - + - ]: 2 : CLOG_W("VDBWriter: Failed to parse stale split manifest [%s]: %s.", path.c_str(), e.what());
339 : 1 : }
340 : : }
341 : :
342 : 33 : impl_->is_split_mode.store(true, std::memory_order_release);
343 : 33 : impl_->split_index.store(0, std::memory_order_relaxed);
344 : :
345 : : #ifdef _WIN32
346 : :
347 : : std::error_code absolute_ec;
348 : : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
349 : :
350 : : if VUNLIKELY (absolute_ec) {
351 : : absolute_path = file_path;
352 : : }
353 : :
354 : : impl_->path = Helpers::path_to_string(absolute_path);
355 : : impl_->split_output_dir = absolute_path.parent_path();
356 : :
357 : : if (parent_path.empty()) {
358 : : impl_->base_dir.clear();
359 : : impl_->base_name = Helpers::path_to_string(file_path.stem());
360 : : } else {
361 : : impl_->base_dir = Helpers::path_to_string(parent_path);
362 : : impl_->base_name = Helpers::path_to_string(std::filesystem::path(parent_path / file_path.stem()));
363 : : }
364 : : #else
365 : :
366 : 33 : std::error_code absolute_ec;
367 [ + - ]: 33 : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
368 : :
369 [ - + ]: 33 : if VUNLIKELY (absolute_ec) {
370 [ # # ]: 0 : absolute_path = file_path;
371 : : }
372 : :
373 [ + - ]: 33 : impl_->path = absolute_path.string();
374 [ + - ]: 33 : impl_->split_output_dir = absolute_path.parent_path();
375 : :
376 [ + + ]: 33 : if (parent_path.empty()) {
377 : 3 : impl_->base_dir.clear();
378 [ + - + - ]: 3 : impl_->base_name = file_path.stem().string();
379 : : } else {
380 [ + - ]: 30 : impl_->base_dir = parent_path.string();
381 [ + - + - : 30 : impl_->base_name = std::filesystem::path(parent_path / file_path.stem()).string();
+ - ]
382 : : }
383 : : #endif
384 : :
385 [ + - ]: 33 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
386 : 33 : impl_->time_current = impl_->time_start;
387 : :
388 [ + + ]: 33 : if (impl_->config.start_timestamp > 0) {
389 : 27 : impl_->start_timestamp = impl_->config.start_timestamp;
390 : : } else {
391 : 6 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
392 : : }
393 : :
394 [ + - ]: 33 : write_filex(false);
395 : :
396 [ + + ]: 33 : if (impl_->config.split_name_by_time) {
397 [ + + ]: 3 : if (impl_->base_dir.empty()) {
398 [ + - + - ]: 2 : impl_->split_filename = get_format_date(&impl_->time_current, true) + ".vdb";
399 : : } else {
400 [ + - + - : 1 : impl_->split_filename = impl_->base_dir + "/" + get_format_date(&impl_->time_current, true) + ".vdb";
+ - + - ]
401 : : }
402 : : } else {
403 : 30 : impl_->split_filename =
404 [ + - + - : 90 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vdb";
+ - + - ]
405 : : }
406 : :
407 [ + - ]: 33 : open_split(impl_->split_filename);
408 : 33 : } else {
409 [ + - ]: 89 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
410 : 89 : impl_->time_current = impl_->time_start;
411 : :
412 [ + + ]: 89 : if (impl_->config.start_timestamp > 0) {
413 : 74 : impl_->start_timestamp = impl_->config.start_timestamp;
414 : : } else {
415 : 15 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
416 : : }
417 : :
418 : 89 : impl_->is_split_mode.store(false, std::memory_order_release);
419 : 89 : impl_->split_index.store(0, std::memory_order_relaxed);
420 : :
421 [ + - ]: 89 : open(path);
422 : : }
423 [ - - ]: 122 : } catch (std::filesystem::filesystem_error& e) {
424 : : VLOG_F("VDBWriter: Filesystem error during init, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
425 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
426 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
427 : :
428 : 122 : impl_->elapsed_timer.start();
429 : 122 : impl_->sync_timer.start();
430 : :
431 : : #ifndef VLINK_ENABLE_SQLITE
432 : : VLOG_F("VDBWriter: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
433 : : #endif
434 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
435 : :
436 : 125 : VDBWriter::~VDBWriter() {
437 : 122 : detach_plugin();
438 : :
439 : 122 : impl_->quit_flag.store(true, std::memory_order_release);
440 : :
441 : 122 : impl_->check_timer.stop();
442 : :
443 [ - + ]: 122 : if VUNLIKELY (!wait_for_idle(30000U)) {
444 : : VLOG_W("VDBWriter: Force to quit."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
445 : : }
446 : :
447 : : #ifdef VLINK_ENABLE_SQLITE
448 : :
449 [ + + ]: 122 : if VLIKELY (impl_->db) {
450 : 103 : ::sqlite3_interrupt(impl_->db);
451 : : }
452 : : #endif
453 : :
454 : 122 : quit(true);
455 : :
456 : 122 : wait_for_quit();
457 : :
458 : 122 : close();
459 : 125 : }
460 : :
461 : 142 : void VDBWriter::close() {
462 : 142 : close_segment();
463 : :
464 [ + + - + : 142 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_acquire) && !write_filex(true)) {
- + ]
465 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
466 : : }
467 : 142 : }
468 : :
469 : 24 : void VDBWriter::register_split_callback(SplitCallback&& callback, bool before) {
470 [ + - ]: 24 : std::lock_guard lock(impl_->split_mtx);
471 : 24 : impl_->split_before = before;
472 : 24 : impl_->split_callback = std::move(callback);
473 : 24 : }
474 : :
475 : 5 : void VDBWriter::register_schema_callback(SchemaCallback&& callback) {
476 [ + - ]: 5 : std::lock_guard lock(impl_->write_mtx);
477 : 5 : impl_->schema_callback = std::move(callback);
478 : 5 : }
479 : :
480 : 41 : bool VDBWriter::merge_schema(SchemaData& schema_data) {
481 : : const auto resolved_schema_type =
482 : 41 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
483 : 41 : schema_data.schema_type = resolved_schema_type;
484 : :
485 [ + + ]: 41 : if VUNLIKELY (schema_data.name.empty()) {
486 : 1 : return true;
487 : : }
488 : :
489 [ + + + + : 40 : if (schema_data.encoding.empty() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
490 [ + - ]: 3 : schema_data.encoding = std::string(SchemaData::convert_type(resolved_schema_type));
491 : : }
492 : :
493 [ + - ]: 40 : std::string schema_key = schema_data.name;
494 [ + - ]: 40 : schema_key.push_back('\x1F');
495 [ + - ]: 40 : schema_key.append(SchemaData::convert_type(resolved_schema_type));
496 : :
497 : 40 : std::string unknown_schema_key;
498 [ + - ]: 40 : auto schema_iter = impl_->total_schema_map.find(schema_key);
499 : :
500 [ + + + + : 40 : if (schema_iter == impl_->total_schema_map.end() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
501 [ + - ]: 36 : unknown_schema_key = schema_data.name;
502 [ + - ]: 36 : unknown_schema_key.push_back('\x1F');
503 [ + - ]: 36 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
504 : : }
505 : :
506 [ + + ]: 40 : if (schema_iter == impl_->total_schema_map.end()) {
507 [ + + + - : 71 : if (!schema_data.encoding.empty() && !schema_data.data.empty() &&
+ - ]
508 [ + - + + ]: 71 : impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
509 [ + - - + ]: 35 : if VUNLIKELY (!insert_schema(schema_data)) {
510 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
511 : : }
512 : :
513 [ + - ]: 35 : impl_->ser_map.emplace(schema_key);
514 : : }
515 : :
516 [ + - ]: 36 : impl_->total_schema_map.emplace(schema_key, schema_data);
517 : :
518 : 36 : return true;
519 : : }
520 : :
521 : 4 : const auto& current = schema_iter->second;
522 : :
523 [ + - + + : 4 : if VUNLIKELY ((!schema_data.encoding.empty() && !current.encoding.empty() &&
+ + - + +
- + - + +
+ + + + +
+ + + + -
+ + + + -
+ - + +
+ ]
524 : : current.encoding != schema_data.encoding) ||
525 : : (!schema_data.data.empty() && !current.data.empty() && current.data != schema_data.data) ||
526 : : (SchemaData::is_real_type(resolved_schema_type) && SchemaData::is_real_type(current.schema_type) &&
527 : : current.schema_type != resolved_schema_type)) {
528 [ + - + - ]: 4 : CLOG_E("VDBWriter: Conflicting schema pushed for [%s].", schema_data.name.c_str());
529 : 2 : return false;
530 : : }
531 : :
532 [ + - ]: 2 : SchemaData merged_schema = current;
533 : :
534 [ + + + - : 2 : if (merged_schema.encoding.empty() && !schema_data.encoding.empty()) {
+ + ]
535 [ + - ]: 1 : merged_schema.encoding = schema_data.encoding;
536 : : }
537 : :
538 [ + + + - : 2 : if (merged_schema.data.empty() && !schema_data.data.empty()) {
+ + ]
539 : 1 : merged_schema.data = schema_data.data;
540 : : }
541 : :
542 [ + + + - : 2 : if (!SchemaData::is_real_type(merged_schema.schema_type) && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
543 : 1 : merged_schema.schema_type = resolved_schema_type;
544 : : }
545 : :
546 [ + - + - : 4 : if (!merged_schema.encoding.empty() && !merged_schema.data.empty() &&
+ + ]
547 [ + - + + ]: 4 : impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
548 [ + - - + ]: 1 : if VUNLIKELY (!insert_schema(merged_schema)) {
549 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
550 : : }
551 : :
552 [ + - ]: 1 : impl_->ser_map.emplace(schema_key);
553 : : }
554 : :
555 [ + - ]: 2 : schema_data = merged_schema;
556 : :
557 [ + + + - : 2 : if (schema_iter->first != schema_key && merged_schema.schema_type == resolved_schema_type) {
+ + ]
558 [ + - ]: 1 : impl_->total_schema_map.erase(schema_iter);
559 [ + - ]: 1 : impl_->total_schema_map.emplace(schema_key, schema_data);
560 : : } else {
561 [ + - ]: 1 : schema_iter->second = schema_data;
562 : : }
563 : :
564 : 2 : return true;
565 : 40 : }
566 : :
567 : 242 : bool VDBWriter::load_schema(const std::string& ser_type, SchemaType& schema_type, SchemaData& schema_data) {
568 : : #ifdef VLINK_ENABLE_SQLITE
569 : 242 : schema_data = SchemaData{};
570 : :
571 [ - + ]: 242 : if VUNLIKELY (ser_type.empty()) {
572 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
573 : : }
574 : :
575 [ + - ]: 242 : std::string schema_key = ser_type;
576 [ + - ]: 242 : schema_key.push_back('\x1F');
577 [ + - ]: 242 : schema_key.append(SchemaData::convert_type(schema_type));
578 : :
579 : 242 : std::string unknown_schema_key;
580 : 242 : auto schema_iter = impl_->total_schema_map.end();
581 : :
582 [ + + ]: 242 : if (schema_type != SchemaType::kUnknown) {
583 [ + - ]: 211 : schema_iter = impl_->total_schema_map.find(schema_key);
584 : :
585 [ + + ]: 211 : if (schema_iter == impl_->total_schema_map.end()) {
586 [ + - ]: 172 : unknown_schema_key = ser_type;
587 [ + - ]: 172 : unknown_schema_key.push_back('\x1F');
588 [ + - ]: 172 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
589 : : }
590 : : } else {
591 [ + - + - ]: 62 : const auto prefix = ser_type + std::string("\x1F");
592 : :
593 [ + + ]: 62 : for (auto iter = impl_->total_schema_map.begin(); iter != impl_->total_schema_map.end(); ++iter) {
594 [ + + ]: 31 : if (!Helpers::has_startwith(iter->first, prefix)) {
595 : 30 : continue;
596 : : }
597 : :
598 [ - + ]: 1 : if (schema_iter != impl_->total_schema_map.end()) {
599 : : schema_iter = impl_->total_schema_map.end(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
600 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
601 : : }
602 : :
603 : 1 : schema_iter = iter;
604 : : }
605 : 31 : }
606 : :
607 [ + + ]: 242 : if VLIKELY (schema_iter != impl_->total_schema_map.end()) {
608 [ + - ]: 40 : schema_data.name = schema_iter->second.name;
609 [ + - ]: 40 : schema_data.encoding = schema_iter->second.encoding;
610 : 40 : schema_data.schema_type = schema_iter->second.schema_type;
611 : 40 : schema_data.data.shallow_copy(schema_iter->second.data);
612 [ - + ]: 202 : } else if (impl_->schema_plugin_interface) {
613 : : schema_data =
614 : : impl_->schema_plugin_interface->search_schema(ser_type, schema_type); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
615 [ + + ]: 202 : } else if (impl_->schema_callback) {
616 [ + - ]: 8 : schema_data = impl_->schema_callback(ser_type, schema_type);
617 : : }
618 : :
619 : 242 : schema_type = SchemaData::resolve_type(schema_type, ser_type, schema_data.encoding);
620 : 242 : schema_data.schema_type = SchemaData::resolve_type(schema_data.schema_type, ser_type, schema_data.encoding);
621 : :
622 [ + + + - ]: 242 : if (schema_type != SchemaType::kUnknown && schema_data.schema_type != SchemaType::kUnknown &&
623 [ + + ]: 212 : schema_type != schema_data.schema_type) {
624 [ + - + - ]: 4 : CLOG_E("VDBWriter: Schema family mismatch for [%s], requested = %d, resolved = %d.", ser_type.c_str(),
625 : : static_cast<int>(schema_type), static_cast<int>(schema_data.schema_type));
626 : 2 : return false;
627 : : }
628 : :
629 [ + + + + : 240 : if (schema_type != SchemaType::kUnknown && schema_data.encoding.empty()) {
+ + ]
630 [ + - ]: 163 : schema_data.encoding = std::string(SchemaData::convert_type(schema_type));
631 : : }
632 : :
633 [ + + - + ]: 240 : if (schema_data.schema_type == SchemaType::kUnknown && schema_type != SchemaType::kUnknown) {
634 : : schema_data.schema_type = schema_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
635 : : }
636 : :
637 [ + + ]: 240 : if (!schema_data.name.empty()) {
638 [ + - ]: 47 : std::string resolved_schema_key = ser_type;
639 [ + - ]: 47 : resolved_schema_key.push_back('\x1F');
640 [ + - ]: 47 : resolved_schema_key.append(SchemaData::convert_type(schema_data.schema_type));
641 : :
642 [ + + + - : 47 : if VLIKELY (schema_iter != impl_->total_schema_map.end() && schema_iter->first == resolved_schema_key) {
+ + ]
643 [ - + - - : 40 : if (schema_iter->second.encoding.empty() && !schema_data.encoding.empty()) {
- + ]
644 [ # # ]: 0 : schema_iter->second.encoding = schema_data.encoding;
645 : : }
646 : :
647 : 40 : schema_iter->second.schema_type = schema_data.schema_type;
648 : : } else {
649 [ + - ]: 7 : SchemaData stored_schema = schema_data;
650 : :
651 [ - + - - : 7 : if (schema_iter != impl_->total_schema_map.end() && schema_iter->second.schema_type == SchemaType::kUnknown &&
- + ]
652 [ # # ]: 0 : schema_data.schema_type != SchemaType::kUnknown) {
653 : : impl_->total_schema_map.erase(schema_iter); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
654 : : }
655 : :
656 : : const auto stored_iter =
657 [ + - ]: 7 : impl_->total_schema_map.insert_or_assign(resolved_schema_key, std::move(stored_schema)).first;
658 : 7 : schema_data.data.shallow_copy(stored_iter->second.data);
659 : 7 : }
660 : 47 : }
661 : :
662 : 240 : return true;
663 : : #else
664 : : (void)ser_type;
665 : : (void)schema_type;
666 : : (void)schema_data;
667 : : return false;
668 : : #endif
669 : 242 : }
670 : :
671 : 41 : bool VDBWriter::push_schema(const SchemaData& schema_data) {
672 [ + - ]: 41 : SchemaData stored_schema = schema_data;
673 : :
674 [ + + ]: 41 : if (impl_->config.sync_mode) {
675 [ + - ]: 38 : std::lock_guard lock(impl_->write_mtx);
676 [ + - ]: 38 : return merge_schema(stored_schema);
677 : 38 : }
678 : :
679 [ + - + - ]: 3 : bool posted = post_persistent_task([this, stored_schema = std::move(stored_schema)]() mutable {
680 [ + - ]: 3 : std::lock_guard lock(impl_->write_mtx);
681 : :
682 [ + - - + ]: 3 : if VUNLIKELY (!merge_schema(stored_schema)) {
683 [ # # # # ]: 0 : CLOG_E("VDBWriter: Deferred merge_schema failed for [%s] in async push_schema path.", stored_schema.name.c_str());
684 : 0 : set_fail();
685 : : }
686 : 3 : });
687 : :
688 : 3 : return posted;
689 : 41 : }
690 : :
691 : 350 : int64_t VDBWriter::record(const Frame& frame, int64_t timestamp) {
692 : : #ifdef VLINK_ENABLE_SQLITE
693 : :
694 : 350 : const std::string& url = frame.url;
695 : 350 : const std::string& ser_type = frame.ser_type;
696 : 350 : const SchemaType schema_type = frame.schema_type;
697 : 350 : const ActionType action_type = frame.action_type;
698 : 350 : const Bytes& data = frame.data;
699 : 350 : const int64_t microseconds_timestamp = timestamp;
700 : :
701 [ + + ]: 350 : if (impl_->config.sync_mode) {
702 [ + - ]: 336 : std::lock_guard lock(impl_->write_mtx);
703 : :
704 [ + - + + ]: 336 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
705 : 6 : return -1;
706 : : }
707 [ + + ]: 336 : } else {
708 [ + + ]: 28 : if VUNLIKELY (impl_->memory_size.load(std::memory_order_relaxed) + static_cast<int64_t>(data.size()) >
709 : : impl_->config.max_memory_size) {
710 [ + - + - ]: 2 : CLOG_E("The memory data in the queue exceeds %.1fGB and the task is automatically discarded.",
711 : : impl_->config.max_memory_size / 1024.0 / 1024.0 / 1024.0);
712 : :
713 : 2 : return -1;
714 : : }
715 : :
716 : 13 : int url_index = -1;
717 : 13 : int ser_index = -1;
718 : :
719 [ + - ]: 13 : get_url_meta(url, ser_type, url_index, ser_index);
720 : :
721 : 13 : const auto queued_size = static_cast<int64_t>(data.size());
722 : :
723 : 13 : impl_->memory_size.fetch_add(queued_size, std::memory_order_relaxed);
724 : 13 : Impl::MemoryCharge memory_charge(impl_->memory_size, queued_size);
725 : :
726 [ + - + - ]: 26 : bool posted = post_persistent_task([this, url_index, ser_index, schema_type, action_type, data,
727 : 13 : memory_charge = std::move(memory_charge),
728 : : microseconds_timestamp]() { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
729 : : (void)memory_charge;
730 : :
731 : 11 : std::string url;
732 : 11 : std::string ser_type;
733 : :
734 [ + - ]: 11 : std::lock_guard lock(impl_->write_mtx);
735 [ + - ]: 11 : get_url_meta(url_index, ser_index, url, ser_type);
736 : :
737 [ + - + + ]: 11 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
738 : 1 : set_fail();
739 : : }
740 : 11 : });
741 : :
742 [ + + ]: 13 : if VUNLIKELY (!posted) {
743 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
744 : : }
745 [ + + ]: 13 : }
746 : :
747 : 342 : return microseconds_timestamp;
748 : : #else
749 : : (void)frame;
750 : : (void)timestamp;
751 : : return -1;
752 : : #endif
753 : : }
754 : :
755 : 1 : int64_t VDBWriter::get_record_timestamp() const { return impl_->elapsed_timer.get(); }
756 : :
757 : 30 : bool VDBWriter::is_dumping() const { return impl_->is_dumping.load(std::memory_order_relaxed); }
758 : :
759 : 60 : bool VDBWriter::is_split_mode() const { return impl_->is_split_mode.load(std::memory_order_acquire); }
760 : :
761 : 140 : int VDBWriter::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
762 : :
763 : 17 : size_t VDBWriter::get_max_task_count() const { return impl_->config.max_task_depth; }
764 : :
765 : 11 : void VDBWriter::on_begin() {
766 : 11 : MessageLoop::on_begin();
767 : :
768 : 11 : impl_->elapsed_timer.restart();
769 : 11 : impl_->sync_timer.restart();
770 : 11 : }
771 : :
772 : 11 : void VDBWriter::on_end() { MessageLoop::on_end(); }
773 : :
774 : 151 : void VDBWriter::open(const std::string& path) {
775 : : #ifdef VLINK_ENABLE_SQLITE
776 : : try {
777 : : #ifdef _WIN32
778 : : impl_->split_file_list.emplace_back(Helpers::path_to_string(std::filesystem::path(path).filename()));
779 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
780 : : #else
781 [ + - + - : 151 : impl_->split_file_list.emplace_back(std::filesystem::path(path).filename().string());
+ - + - ]
782 [ + - ]: 151 : std::filesystem::path file_path(path);
783 : : #endif
784 : :
785 [ + - + + ]: 151 : if (std::filesystem::exists(file_path)) {
786 [ + - ]: 1 : std::filesystem::remove(file_path);
787 : : } else {
788 [ + - ]: 150 : auto parent_path = file_path.parent_path();
789 : :
790 [ + - + - : 150 : if (!parent_path.empty() && !std::filesystem::exists(parent_path)) {
+ + + + ]
791 [ + - ]: 1 : std::filesystem::create_directories(parent_path);
792 : : }
793 : 150 : }
794 [ - - ]: 151 : } catch (std::filesystem::filesystem_error& e) {
795 : : VLOG_F("VDBWriter: Filesystem error during file preparation, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
796 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
797 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
798 : :
799 : 151 : int ret = 0;
800 : 151 : char* err_msg = nullptr;
801 : :
802 : 151 : auto free_err_msg = [&err_msg]() noexcept {
803 [ - + ]: 151 : if (err_msg) {
804 : : ::sqlite3_free(err_msg); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
805 : : err_msg = nullptr; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
806 : : }
807 : 302 : };
808 : :
809 : 151 : auto finalize_stmt = [](sqlite3_stmt*& stmt) noexcept {
810 [ + - ]: 151 : if (stmt) {
811 : 151 : ::sqlite3_finalize(stmt);
812 : 151 : stmt = nullptr;
813 : : }
814 : 151 : };
815 : :
816 : 0 : auto close_db = [this, &finalize_stmt]() noexcept {
817 : : // LCOV_EXCL_START GCOVR_EXCL_START
818 : : if (!impl_->db) {
819 : : return;
820 : : }
821 : :
822 : : if (impl_->in_cached.load(std::memory_order_relaxed)) {
823 : : ::sqlite3_exec(impl_->db, "ROLLBACK;", nullptr, nullptr, nullptr);
824 : : impl_->in_cached.store(false, std::memory_order_relaxed);
825 : : impl_->cached_size.store(0, std::memory_order_relaxed);
826 : : impl_->cache_timer.stop();
827 : : }
828 : :
829 : : finalize_stmt(impl_->schemas_stmt);
830 : : finalize_stmt(impl_->datas_stmt);
831 : : finalize_stmt(impl_->urls_stmt);
832 : : finalize_stmt(impl_->update_complete_stmt);
833 : : finalize_stmt(impl_->update_header_stmt);
834 : : finalize_stmt(impl_->update_url_loss_stmt);
835 : : finalize_stmt(impl_->update_url_meta_stmt);
836 : : finalize_stmt(impl_->update_urls_stmt);
837 : :
838 : : const char* err_ptr = ::sqlite3_errmsg(impl_->db);
839 : : std::string close_err = err_ptr ? err_ptr : std::string{};
840 : : int close_ret = ::sqlite3_close_v2(impl_->db);
841 : :
842 : : if VUNLIKELY (close_ret != SQLITE_OK) {
843 : : CLOG_W("Failed to close database (rc=%d): %s.", close_ret, close_err.c_str());
844 : : }
845 : :
846 : : impl_->db = nullptr;
847 : : };
848 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
849 : :
850 : : // open db
851 [ + - ]: 151 : ret = ::sqlite3_open_v2(path.c_str(), &impl_->db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, nullptr);
852 : :
853 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
854 : : CLOG_F("Failed to open database [%s].", path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
855 : : close_db(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
856 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
857 : : }
858 : :
859 : : // opt busy_timeout
860 [ + - ]: 151 : ::sqlite3_busy_timeout(impl_->db, 100);
861 : :
862 : : // opt sqlite temp_store
863 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA temp_store = MEMORY;", nullptr, nullptr, &err_msg);
864 : :
865 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
866 : : // LCOV_EXCL_START GCOVR_EXCL_START
867 : : CLOG_F("Failed to set temp_store: %s.", err_msg);
868 : :
869 : : free_err_msg();
870 : : close_db();
871 : :
872 : : return;
873 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
874 : : }
875 : :
876 : : // opt sqlite page_size
877 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA page_size = 16384;", nullptr, nullptr, &err_msg);
878 : :
879 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
880 : : // LCOV_EXCL_START GCOVR_EXCL_START
881 : : CLOG_F("Failed to set page_size: %s.", err_msg);
882 : :
883 : : free_err_msg();
884 : : close_db();
885 : :
886 : : return;
887 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
888 : : }
889 : :
890 : : // opt sqlite cache_size
891 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA cache_size = 8192;", nullptr, nullptr, &err_msg);
892 : :
893 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
894 : : // LCOV_EXCL_START GCOVR_EXCL_START
895 : : CLOG_F("Failed to set cache_size: %s.", err_msg);
896 : :
897 : : free_err_msg();
898 : : close_db();
899 : :
900 : : return;
901 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
902 : : }
903 : :
904 : : // opt sqlite synchronous
905 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA synchronous = OFF;", nullptr, nullptr, &err_msg);
906 : :
907 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
908 : : // LCOV_EXCL_START GCOVR_EXCL_START
909 : : CLOG_F("Failed to set synchronous: %s.", err_msg);
910 : :
911 : : free_err_msg();
912 : : close_db();
913 : :
914 : : return;
915 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
916 : : }
917 : :
918 : : // opt sqlite journal_mode
919 : :
920 [ + + ]: 151 : if (impl_->config.wal_mode) {
921 [ + - ]: 2 : ret = ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = WAL;", nullptr, nullptr, &err_msg);
922 [ - + ]: 2 : if VUNLIKELY (ret != SQLITE_OK) {
923 : : // LCOV_EXCL_START GCOVR_EXCL_START
924 : : CLOG_F("Failed to set journal_mode: %s.", err_msg);
925 : :
926 : : free_err_msg();
927 : : close_db();
928 : :
929 : : return;
930 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
931 : : }
932 : : } else {
933 [ + - ]: 149 : ret = ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = OFF;", nullptr, nullptr, &err_msg);
934 [ - + ]: 149 : if VUNLIKELY (ret != SQLITE_OK) {
935 : : // LCOV_EXCL_START GCOVR_EXCL_START
936 : : CLOG_F("Failed to set journal_mode: %s.", err_msg);
937 : :
938 : : free_err_msg();
939 : : close_db();
940 : :
941 : : return;
942 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
943 : : }
944 : : }
945 : :
946 : : // opt sqlite automatic_index
947 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA automatic_index = OFF;", nullptr, nullptr, &err_msg);
948 : :
949 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
950 : : // LCOV_EXCL_START GCOVR_EXCL_START
951 : : CLOG_F("Failed to set automatic_index: %s.", err_msg);
952 : :
953 : : free_err_msg();
954 : : close_db();
955 : :
956 : : return;
957 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
958 : : }
959 : :
960 : : // opt sqlite locking_mode
961 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "PRAGMA locking_mode = EXCLUSIVE;", nullptr, nullptr, &err_msg);
962 : :
963 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
964 : : // LCOV_EXCL_START GCOVR_EXCL_START
965 : : CLOG_F("Failed to set locking_mode: %s.", err_msg);
966 : :
967 : : free_err_msg();
968 : : close_db();
969 : :
970 : : return;
971 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
972 : : }
973 : :
974 : : // set busy_timeout
975 [ + - ]: 151 : ::sqlite3_busy_timeout(impl_->db, 500);
976 : :
977 [ + - ]: 151 : begin_cache();
978 : :
979 : : // create header table
980 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db,
981 : : "CREATE TABLE IF NOT EXISTS VLinkHeader(major INTEGER, minor INTEGER, patch INTEGER, "
982 : : "count INTEGER, duration INTEGER, accuracy TEXT, compress TEXT, process TEXT, date TEXT, "
983 : : "tag TEXT, complete INTEGER, timezone INTEGER, start_timestamp INTEGER);",
984 : : nullptr, nullptr, &err_msg);
985 : :
986 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
987 : : // LCOV_EXCL_START GCOVR_EXCL_START
988 : : CLOG_F("Failed to create header table: %s.", err_msg);
989 : :
990 : : free_err_msg();
991 : : close_db();
992 : :
993 : : return;
994 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
995 : : }
996 : :
997 : : // prepare header table
998 : 151 : ::sqlite3_stmt* header_stmt = nullptr;
999 [ + - ]: 151 : ret = ::sqlite3_prepare_v2(
1000 : 151 : impl_->db,
1001 : : "INSERT INTO VLinkHeader (major, minor, patch, count, duration, accuracy, compress, "
1002 : : "process, date, tag, complete, timezone, start_timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
1003 : : -1, &header_stmt, nullptr);
1004 : :
1005 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1006 : : // LCOV_EXCL_START GCOVR_EXCL_START
1007 : : CLOG_F("Failed to prepare header table: %s.", ::sqlite3_errmsg(impl_->db));
1008 : :
1009 : : free_err_msg();
1010 : : finalize_stmt(header_stmt);
1011 : : close_db();
1012 : :
1013 : : return;
1014 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1015 : : }
1016 : :
1017 : : // get date
1018 [ + - ]: 151 : const std::string& date_str = get_format_date(&impl_->time_current);
1019 : :
1020 : : // insert header table
1021 [ + - ]: 151 : ::sqlite3_bind_int(header_stmt, get_column(0), VLINK_VERSION_MAJOR);
1022 [ + - ]: 151 : ::sqlite3_bind_int(header_stmt, get_column(1), VLINK_VERSION_MINOR);
1023 [ + - ]: 151 : ::sqlite3_bind_int(header_stmt, get_column(2), VLINK_VERSION_PATCH);
1024 [ + - ]: 151 : ::sqlite3_bind_int64(header_stmt, get_column(3), 0);
1025 [ + - ]: 151 : ::sqlite3_bind_int64(header_stmt, get_column(4), 0);
1026 [ + - ]: 151 : ::sqlite3_bind_text(header_stmt, get_column(5), "MicroSecond", -1, SQLITE_STATIC);
1027 [ + + + - ]: 151 : ::sqlite3_bind_text(header_stmt, get_column(6), impl_->enable_compressed ? "lzav" : "None", -1, SQLITE_STATIC);
1028 [ + - ]: 151 : ::sqlite3_bind_text(header_stmt, get_column(7), impl_->app_name.data(), impl_->app_name.size(), SQLITE_STATIC);
1029 [ + - ]: 151 : ::sqlite3_bind_text(header_stmt, get_column(8), date_str.data(), date_str.size(), SQLITE_STATIC);
1030 [ + - ]: 151 : ::sqlite3_bind_text(header_stmt, get_column(9), impl_->tag_name.data(), impl_->tag_name.size(), SQLITE_STATIC);
1031 [ + - ]: 151 : ::sqlite3_bind_int(header_stmt, get_column(10), 0);
1032 [ + - ]: 151 : ::sqlite3_bind_int(header_stmt, get_column(11), impl_->timezone_diff);
1033 [ + - ]: 151 : ::sqlite3_bind_int64(header_stmt, get_column(12), impl_->start_timestamp);
1034 : :
1035 [ + - ]: 151 : ret = ::sqlite3_step(header_stmt);
1036 : :
1037 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_DONE) {
1038 : : // LCOV_EXCL_START GCOVR_EXCL_START
1039 : : CLOG_F("Failed to insert header table: %s.", ::sqlite3_errmsg(impl_->db));
1040 : :
1041 : : free_err_msg();
1042 : : finalize_stmt(header_stmt);
1043 : : close_db();
1044 : :
1045 : : return;
1046 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1047 : : }
1048 : :
1049 : 151 : finalize_stmt(header_stmt);
1050 : :
1051 : : // create schema table
1052 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "CREATE TABLE IF NOT EXISTS VLinkSchemas(ser TEXT, encoding TEXT, data BLOB);",
1053 : : nullptr, nullptr, &err_msg);
1054 : :
1055 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1056 : : // LCOV_EXCL_START GCOVR_EXCL_START
1057 : : CLOG_F("Failed to create schema table: %s.", err_msg);
1058 : :
1059 : : free_err_msg();
1060 : : close_db();
1061 : :
1062 : : return;
1063 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1064 : : }
1065 : :
1066 : : // prepare schema table
1067 [ + - ]: 151 : ret = ::sqlite3_prepare_v2(impl_->db, "INSERT INTO VLinkSchemas(ser, encoding, data) VALUES (?, ?, ?);", -1,
1068 : 151 : &impl_->schemas_stmt, nullptr);
1069 : :
1070 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1071 : : // LCOV_EXCL_START GCOVR_EXCL_START
1072 : : CLOG_F("Failed to prepare schema table: %s.", ::sqlite3_errmsg(impl_->db));
1073 : :
1074 : : free_err_msg();
1075 : : close_db();
1076 : :
1077 : : return;
1078 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1079 : : }
1080 : :
1081 : : // create url table
1082 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db,
1083 : : "CREATE TABLE IF NOT EXISTS VLinkUrls(id INTEGER, url TEXT, type TEXT, ser TEXT, encoding TEXT, "
1084 : : "count INTEGER, loss REAL, size INTEGER, freq REAL);",
1085 : : nullptr, nullptr, &err_msg);
1086 : :
1087 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1088 : : // LCOV_EXCL_START GCOVR_EXCL_START
1089 : : CLOG_F("Failed to create urls table: %s.", err_msg);
1090 : :
1091 : : free_err_msg();
1092 : : close_db();
1093 : :
1094 : : return;
1095 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1096 : : }
1097 : :
1098 : : // prepare url table
1099 [ + - ]: 151 : ret = ::sqlite3_prepare_v2(impl_->db,
1100 : : "INSERT INTO VLinkUrls (id, url, type, ser, encoding, count, loss, size, freq) VALUES "
1101 : : "(?, ?, ?, ?, ?, ?, ?, ?, ?);",
1102 : 151 : -1, &impl_->urls_stmt, nullptr);
1103 : :
1104 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1105 : : // LCOV_EXCL_START GCOVR_EXCL_START
1106 : : CLOG_F("Failed to prepare urls table: %s.", ::sqlite3_errmsg(impl_->db));
1107 : :
1108 : : free_err_msg();
1109 : : close_db();
1110 : :
1111 : : return;
1112 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1113 : : }
1114 : :
1115 : : // create data table
1116 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db,
1117 : : "CREATE TABLE IF NOT EXISTS VLinkDatas(elapsed INTEGER, url INTEGER, action TEXT, data BLOB);",
1118 : : nullptr, nullptr, &err_msg);
1119 : :
1120 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1121 : : // LCOV_EXCL_START GCOVR_EXCL_START
1122 : : CLOG_F("Failed to create datas table: %s.", err_msg);
1123 : :
1124 : : free_err_msg();
1125 : : close_db();
1126 : :
1127 : : return;
1128 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1129 : : }
1130 : :
1131 : : // prepare data table
1132 [ + - ]: 151 : ret = ::sqlite3_prepare_v2(impl_->db, "INSERT INTO VLinkDatas (elapsed, url, action, data) VALUES (?, ?, ?, ?);", -1,
1133 : 151 : &impl_->datas_stmt, nullptr);
1134 : :
1135 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1136 : : // LCOV_EXCL_START GCOVR_EXCL_START
1137 : : CLOG_F("Failed to prepare datas table: %s.", ::sqlite3_errmsg(impl_->db));
1138 : :
1139 : : free_err_msg();
1140 : : close_db();
1141 : :
1142 : : return;
1143 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1144 : : }
1145 : :
1146 : : // prepare update_header table
1147 [ + - ]: 151 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkHeader SET count=?, duration=? WHERE ROWID=1;", -1,
1148 : 151 : &impl_->update_header_stmt, nullptr);
1149 : :
1150 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1151 : : // LCOV_EXCL_START GCOVR_EXCL_START
1152 : : CLOG_F("Failed to prepare update_header: %s.", ::sqlite3_errmsg(impl_->db));
1153 : :
1154 : : free_err_msg();
1155 : : close_db();
1156 : :
1157 : : return;
1158 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1159 : : }
1160 : :
1161 : : // prepare update_url table
1162 [ + - ]: 151 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET count=?, size=?, freq=? WHERE url=?;", -1,
1163 : 151 : &impl_->update_urls_stmt, nullptr);
1164 : :
1165 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1166 : : // LCOV_EXCL_START GCOVR_EXCL_START
1167 : : CLOG_F("Failed to prepare update_urls: %s.", ::sqlite3_errmsg(impl_->db));
1168 : :
1169 : : free_err_msg();
1170 : : close_db();
1171 : :
1172 : : return;
1173 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1174 : : }
1175 : :
1176 : : // create idx_elapsed_url
1177 [ + - ]: 151 : ret = ::sqlite3_exec(impl_->db, "CREATE INDEX IF NOT EXISTS idx_elapsed_url ON VLinkDatas(elapsed, url);", nullptr,
1178 : : nullptr, &err_msg);
1179 : :
1180 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1181 : : // LCOV_EXCL_START GCOVR_EXCL_START
1182 : : CLOG_F("Failed to create datas idx_elapsed_url: %s.", err_msg);
1183 : :
1184 : : free_err_msg();
1185 : : close_db();
1186 : :
1187 : : return;
1188 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1189 : : }
1190 : :
1191 [ + - ]: 151 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET ser=?, encoding=? WHERE url=?;", -1,
1192 : 151 : &impl_->update_url_meta_stmt, nullptr);
1193 : :
1194 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1195 : : // LCOV_EXCL_START GCOVR_EXCL_START
1196 : : CLOG_F("Failed to prepare update_url_meta: %s.", ::sqlite3_errmsg(impl_->db));
1197 : :
1198 : : free_err_msg();
1199 : : close_db();
1200 : :
1201 : : return;
1202 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1203 : : }
1204 : :
1205 [ + - ]: 151 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET loss=? WHERE url=?;", -1, &impl_->update_url_loss_stmt,
1206 : : nullptr);
1207 : :
1208 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1209 : : // LCOV_EXCL_START GCOVR_EXCL_START
1210 : : CLOG_F("Failed to prepare update_url_loss: %s.", ::sqlite3_errmsg(impl_->db));
1211 : :
1212 : : free_err_msg();
1213 : : close_db();
1214 : :
1215 : : return;
1216 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1217 : : }
1218 : :
1219 [ + - ]: 151 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkHeader SET complete=1 WHERE ROWID=1;", -1,
1220 : 151 : &impl_->update_complete_stmt, nullptr);
1221 : :
1222 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1223 : : // LCOV_EXCL_START GCOVR_EXCL_START
1224 : : CLOG_F("Failed to prepare update_complete: %s.", ::sqlite3_errmsg(impl_->db));
1225 : :
1226 : : free_err_msg();
1227 : : close_db();
1228 : :
1229 : : return;
1230 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1231 : : }
1232 : :
1233 : 151 : free_err_msg();
1234 : :
1235 [ + - ]: 151 : sync_cache();
1236 : :
1237 : : #else
1238 : : (void)path;
1239 : : VLOG_F("VDBWriter: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
1240 : : #endif
1241 : :
1242 : 151 : impl_->last_timestamp = 0;
1243 [ + - ]: 151 : }
1244 : :
1245 : 62 : void VDBWriter::open_split(const std::string& path) {
1246 : : #ifdef _WIN32
1247 : : const auto file_name = std::filesystem::path(Helpers::string_to_wstring(path)).filename();
1248 : : open(Helpers::path_to_string(impl_->split_output_dir / file_name));
1249 : : #else
1250 [ + - + - : 62 : open((impl_->split_output_dir / std::filesystem::path(path).filename()).string());
+ - + - ]
1251 : : #endif
1252 : 62 : }
1253 : :
1254 : 171 : void VDBWriter::close_segment() {
1255 : : #ifdef VLINK_ENABLE_SQLITE
1256 : :
1257 [ + + ]: 171 : if (!impl_->db) {
1258 : 20 : return;
1259 : : }
1260 : :
1261 : 151 : bool close_success = sync_cache();
1262 : :
1263 [ - + ]: 151 : if (!close_success) {
1264 : : if (impl_->in_cached.load(std::memory_order_relaxed)) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1265 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1266 : : }
1267 : : } else {
1268 : 151 : close_success = begin_cache();
1269 : : }
1270 : :
1271 [ + - ]: 151 : if (close_success) {
1272 : 151 : int ret = SQLITE_OK;
1273 : :
1274 [ + + ]: 359 : for (const auto& [url, info] : impl_->url_map) {
1275 [ + - ]: 208 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(0), info.count);
1276 [ + - ]: 208 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(1), info.size);
1277 [ + - ]: 208 : ::sqlite3_bind_double(impl_->update_urls_stmt, get_column(2), info.freq);
1278 [ + - ]: 208 : ::sqlite3_bind_text(impl_->update_urls_stmt, get_column(3), url.c_str(), url.size(), SQLITE_STATIC);
1279 : :
1280 [ + - ]: 208 : ret = ::sqlite3_step(impl_->update_urls_stmt);
1281 : :
1282 [ - + ]: 208 : if VUNLIKELY (ret != SQLITE_DONE) {
1283 : : CLOG_W("Failed to update urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1284 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1285 : : }
1286 : :
1287 [ + - ]: 208 : ret = ::sqlite3_reset(impl_->update_urls_stmt);
1288 : :
1289 [ - + ]: 208 : if VUNLIKELY (ret != SQLITE_OK) {
1290 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1291 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1292 : : }
1293 : :
1294 [ - + ]: 208 : if VUNLIKELY (!close_success) {
1295 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1296 : : }
1297 : : }
1298 : :
1299 [ + - ]: 151 : if (close_success) {
1300 : 151 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(0), impl_->current_row);
1301 : 151 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(1), impl_->last_timestamp);
1302 : :
1303 : 151 : ret = ::sqlite3_step(impl_->update_header_stmt);
1304 : :
1305 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_DONE) {
1306 : : CLOG_W("Failed to update header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1307 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1308 : : }
1309 : :
1310 : 151 : ret = ::sqlite3_reset(impl_->update_header_stmt);
1311 : :
1312 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1313 : : CLOG_W("Failed to reset header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1314 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1315 : : }
1316 : : }
1317 : :
1318 [ + - ]: 151 : if (close_success) {
1319 [ + - + - ]: 151 : std::lock_guard lock(sample_mutex());
1320 : :
1321 [ + - + + ]: 190 : for (const auto& [url, loss] : url_loss_map_ref()) {
1322 [ + - + + ]: 78 : for (const auto& recorded_url : recorded_urls_for_origin(url)) {
1323 [ + - ]: 39 : ::sqlite3_bind_double(impl_->update_url_loss_stmt, get_column(0), loss);
1324 [ + - ]: 39 : ::sqlite3_bind_text(impl_->update_url_loss_stmt, get_column(1), recorded_url.c_str(), recorded_url.size(),
1325 : : SQLITE_STATIC);
1326 : :
1327 [ + - ]: 39 : ret = ::sqlite3_step(impl_->update_url_loss_stmt);
1328 : :
1329 [ - + ]: 39 : if VUNLIKELY (ret != SQLITE_DONE) {
1330 : : CLOG_W("Failed to update url loss: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1331 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1332 : : }
1333 : :
1334 [ + - ]: 39 : ret = ::sqlite3_reset(impl_->update_url_loss_stmt);
1335 : :
1336 [ - + ]: 39 : if VUNLIKELY (ret != SQLITE_OK) {
1337 : : CLOG_W("Failed to reset url loss: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1338 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1339 : : }
1340 : :
1341 [ - + ]: 39 : if VUNLIKELY (!close_success) {
1342 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1343 : : }
1344 : 39 : }
1345 : :
1346 [ - + ]: 39 : if VUNLIKELY (!close_success) {
1347 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1348 : : }
1349 : : }
1350 : 151 : }
1351 : :
1352 [ + - ]: 151 : if (close_success) {
1353 : 151 : ret = ::sqlite3_step(impl_->update_complete_stmt);
1354 : :
1355 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_DONE) {
1356 : : CLOG_W("Failed to mark database complete: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1357 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1358 : : }
1359 : :
1360 : 151 : ret = ::sqlite3_reset(impl_->update_complete_stmt);
1361 : :
1362 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1363 : : CLOG_W("Failed to reset complete flag: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1364 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1365 : : }
1366 : : }
1367 : :
1368 [ + - ]: 151 : if (close_success) {
1369 : 151 : close_success = sync_cache();
1370 : : }
1371 : : }
1372 : :
1373 [ - + ]: 151 : if VUNLIKELY (!close_success) {
1374 : 0 : set_fail();
1375 : :
1376 [ # # ]: 0 : if (impl_->in_cached.load(std::memory_order_relaxed)) {
1377 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1378 : : }
1379 : : }
1380 : :
1381 [ + - ]: 151 : if VLIKELY (impl_->datas_stmt) {
1382 : 151 : ::sqlite3_finalize(impl_->datas_stmt);
1383 : 151 : impl_->datas_stmt = nullptr;
1384 : : }
1385 : :
1386 [ + - ]: 151 : if VLIKELY (impl_->urls_stmt) {
1387 : 151 : ::sqlite3_finalize(impl_->urls_stmt);
1388 : 151 : impl_->urls_stmt = nullptr;
1389 : : }
1390 : :
1391 [ + - ]: 151 : if VLIKELY (impl_->schemas_stmt) {
1392 : 151 : ::sqlite3_finalize(impl_->schemas_stmt);
1393 : 151 : impl_->schemas_stmt = nullptr;
1394 : : }
1395 : :
1396 [ + - ]: 151 : if VLIKELY (impl_->update_urls_stmt) {
1397 : 151 : ::sqlite3_finalize(impl_->update_urls_stmt);
1398 : 151 : impl_->update_urls_stmt = nullptr;
1399 : : }
1400 : :
1401 [ + - ]: 151 : if VLIKELY (impl_->update_url_meta_stmt) {
1402 : 151 : ::sqlite3_finalize(impl_->update_url_meta_stmt);
1403 : 151 : impl_->update_url_meta_stmt = nullptr;
1404 : : }
1405 : :
1406 [ + - ]: 151 : if VLIKELY (impl_->update_url_loss_stmt) {
1407 : 151 : ::sqlite3_finalize(impl_->update_url_loss_stmt);
1408 : 151 : impl_->update_url_loss_stmt = nullptr;
1409 : : }
1410 : :
1411 [ + - ]: 151 : if VLIKELY (impl_->update_header_stmt) {
1412 : 151 : ::sqlite3_finalize(impl_->update_header_stmt);
1413 : 151 : impl_->update_header_stmt = nullptr;
1414 : : }
1415 : :
1416 [ + - ]: 151 : if VLIKELY (impl_->update_complete_stmt) {
1417 : 151 : ::sqlite3_finalize(impl_->update_complete_stmt);
1418 : 151 : impl_->update_complete_stmt = nullptr;
1419 : : }
1420 : :
1421 [ + + ]: 151 : if (impl_->config.wal_mode) {
1422 : 2 : ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = OFF;", nullptr, nullptr, nullptr);
1423 : : }
1424 : :
1425 [ + + ]: 151 : if (impl_->config.optimize_on_exit) {
1426 : 18 : ::sqlite3_exec(impl_->db, "PRAGMA optimize;", nullptr, nullptr, nullptr);
1427 : : }
1428 : :
1429 [ + - ]: 151 : if VLIKELY (impl_->db) {
1430 [ + - ]: 151 : const char* err_ptr = ::sqlite3_errmsg(impl_->db);
1431 [ + - + - : 151 : std::string close_err = err_ptr ? err_ptr : std::string{};
+ - - - ]
1432 [ + - ]: 151 : int ret = ::sqlite3_close_v2(impl_->db);
1433 : :
1434 [ - + ]: 151 : if VUNLIKELY (ret != SQLITE_OK) {
1435 : : CLOG_W("Failed to close database (rc=%d): %s.", ret, close_err.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1436 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1437 : : }
1438 : :
1439 : 151 : impl_->db = nullptr;
1440 : 151 : }
1441 : :
1442 : 151 : impl_->url_map.clear();
1443 : 151 : impl_->ser_map.clear();
1444 : 151 : url_loss_map_ref().clear();
1445 : :
1446 : 151 : impl_->current_row = 0;
1447 : 151 : impl_->current_size = 0;
1448 : 151 : impl_->has_oversize = false;
1449 : :
1450 : 151 : impl_->in_cached.store(false, std::memory_order_relaxed);
1451 : 151 : impl_->cached_size.store(0, std::memory_order_relaxed);
1452 : 151 : impl_->cache_snapshot.reset();
1453 : : #endif
1454 : :
1455 : 151 : impl_->last_timestamp = 0;
1456 : : }
1457 : :
1458 : 347 : bool VDBWriter::write(const std::string& url, const std::string& ser_type, SchemaType schema_type,
1459 : : ActionType action_type, const Bytes& data, int64_t microseconds_timestamp) {
1460 : : #ifdef VLINK_ENABLE_SQLITE
1461 : :
1462 [ - + ]: 347 : if VUNLIKELY (!impl_->db) {
1463 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1464 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1465 : : }
1466 : :
1467 [ + + + + : 347 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_acquire) && !impl_->split_first) {
+ + ]
1468 : 32 : impl_->split_first = true;
1469 : :
1470 [ + - ]: 32 : std::lock_guard split_lock(impl_->split_mtx);
1471 : :
1472 [ + + + + : 35 : if (!impl_->split_before && impl_->split_callback && impl_->split_index.load(std::memory_order_relaxed) == 0) {
+ - + + ]
1473 [ + - ]: 3 : impl_->split_callback(0, impl_->split_filename);
1474 : : }
1475 : 32 : }
1476 : :
1477 : 347 : int ret = 0;
1478 : :
1479 : 347 : bool do_compress = false;
1480 : 347 : Bytes compressed_data;
1481 : 347 : bool do_split = false;
1482 : :
1483 [ + - - + ]: 347 : if VUNLIKELY (!begin_cache()) {
1484 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1485 : : }
1486 : :
1487 : : // delete data, when limit
1488 [ + + + + : 355 : while (impl_->current_row > impl_->config.max_row_count || impl_->current_size > impl_->config.max_bytes_size) {
+ + ]
1489 [ + + ]: 10 : if VUNLIKELY (!impl_->has_oversize) {
1490 [ + + ]: 8 : if (impl_->config.enable_limit) {
1491 [ + - + - ]: 12 : VLOG_W("VDBWriter: The number of messages has reached the upper limit, the oldest data will be deleted.");
1492 : : } else {
1493 [ + - + - ]: 4 : VLOG_W("VDBWriter: The number of messages has reached the upper limit, data after that will be ignored.");
1494 : : }
1495 : :
1496 : 8 : impl_->has_oversize = true;
1497 : : }
1498 : :
1499 [ + + ]: 10 : if (!impl_->config.enable_limit) {
1500 [ + - ]: 2 : rollback_cache();
1501 : 2 : return false;
1502 : : }
1503 : :
1504 : 8 : ::sqlite3_stmt* select_stmt = nullptr;
1505 : 8 : ::sqlite3_stmt* delete_row_stmt = nullptr;
1506 [ + - ]: 8 : ret = ::sqlite3_prepare_v2(impl_->db, "SELECT rowid, url, length(data) FROM VLinkDatas ORDER BY rowid LIMIT 1;", -1,
1507 : : &select_stmt, nullptr);
1508 : :
1509 [ + - ]: 8 : if VLIKELY (ret == SQLITE_OK) {
1510 [ + - ]: 8 : ret = ::sqlite3_step(select_stmt);
1511 : :
1512 [ + - ]: 8 : if VLIKELY (ret == SQLITE_ROW) {
1513 [ + - ]: 8 : const auto erase_row_id = ::sqlite3_column_int64(select_stmt, 0);
1514 [ + - ]: 8 : const auto erase_url_id = ::sqlite3_column_int(select_stmt, 1);
1515 [ + - ]: 8 : const auto erase_blob_size = ::sqlite3_column_int64(select_stmt, 2);
1516 : 8 : auto erase_size = static_cast<size_t>(erase_blob_size);
1517 : :
1518 [ + + + - : 8 : if (impl_->enable_compressed && erase_blob_size >= 13) {
+ + ]
1519 : 6 : std::array<uint8_t, 13> erase_data{};
1520 : 6 : ::sqlite3_blob* erase_blob = nullptr;
1521 [ + - ]: 6 : ret = ::sqlite3_blob_open(impl_->db, "main", "VLinkDatas", "data", erase_row_id, 0, &erase_blob);
1522 : :
1523 [ + - ]: 6 : if VLIKELY (ret == SQLITE_OK) {
1524 [ + - ]: 6 : ret = ::sqlite3_blob_read(erase_blob, erase_data.data(), 9, 0);
1525 : : }
1526 : :
1527 [ + - ]: 6 : if VLIKELY (ret == SQLITE_OK) {
1528 [ + - ]: 6 : ret = ::sqlite3_blob_read(erase_blob, erase_data.data() + 9, 4, static_cast<int>(erase_blob_size) - 4);
1529 : : }
1530 : :
1531 [ + - ]: 6 : if (erase_blob) {
1532 [ + - ]: 6 : const auto close_ret = ::sqlite3_blob_close(erase_blob);
1533 [ + - ]: 6 : if (ret == SQLITE_OK) {
1534 : 6 : ret = close_ret;
1535 : : }
1536 : : }
1537 : :
1538 [ - + ]: 6 : if VUNLIKELY (ret != SQLITE_OK) {
1539 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to read compressed data metadata while evicting: %s.",
# # ]
1540 : : ::sqlite3_errmsg(impl_->db));
1541 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1542 [ # # ]: 0 : rollback_cache();
1543 : 0 : return false;
1544 : : }
1545 : :
1546 [ + - ]: 6 : if (Bytes::is_compress_data(erase_data.data(), erase_data.size())) {
1547 : 6 : erase_size = (static_cast<size_t>(erase_data[4]) << 24U) | (static_cast<size_t>(erase_data[5]) << 16U) |
1548 : 6 : (static_cast<size_t>(erase_data[6]) << 8U) | static_cast<size_t>(erase_data[7]);
1549 : : }
1550 : : }
1551 : :
1552 : 8 : auto total_url_iter = impl_->total_url_map.end();
1553 : :
1554 [ + - + - : 8 : if VLIKELY (erase_url_id >= 0 && static_cast<size_t>(erase_url_id) < impl_->total_url_list.size()) {
+ - ]
1555 [ + - ]: 8 : total_url_iter = impl_->total_url_map.find(impl_->total_url_list[static_cast<size_t>(erase_url_id)]);
1556 : : }
1557 : :
1558 [ - + ]: 8 : if VUNLIKELY (total_url_iter == impl_->total_url_map.end()) {
1559 [ # # # # ]: 0 : CLOG_W("VDBWriter: URL metadata not found while evicting the oldest data.");
1560 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1561 [ # # ]: 0 : rollback_cache();
1562 : 0 : return false;
1563 : : }
1564 : :
1565 [ + - ]: 8 : auto url_iter = impl_->url_map.find(total_url_iter->first);
1566 : :
1567 [ - + ]: 8 : if VUNLIKELY (url_iter == impl_->url_map.end()) {
1568 [ # # # # ]: 0 : CLOG_W("VDBWriter: Segment URL metadata not found while evicting the oldest data.");
1569 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1570 [ # # ]: 0 : rollback_cache();
1571 : 0 : return false;
1572 : : }
1573 : :
1574 [ + - ]: 8 : ::sqlite3_finalize(select_stmt);
1575 : 8 : select_stmt = nullptr;
1576 : :
1577 [ + - ]: 8 : ret = ::sqlite3_prepare_v2(impl_->db, "DELETE FROM VLinkDatas WHERE rowid = ?;", -1, &delete_row_stmt, nullptr);
1578 : :
1579 [ + - ]: 8 : if VLIKELY (ret == SQLITE_OK) {
1580 [ + - ]: 8 : ::sqlite3_bind_int64(delete_row_stmt, get_column(0), erase_row_id);
1581 [ + - ]: 8 : ret = ::sqlite3_step(delete_row_stmt);
1582 : : }
1583 : :
1584 [ + - ]: 8 : if VLIKELY (ret == SQLITE_DONE) {
1585 : 8 : --impl_->current_row;
1586 : 8 : impl_->current_size -= static_cast<int64_t>(erase_size);
1587 : 8 : --impl_->total_current_row;
1588 : 8 : impl_->total_current_size -= static_cast<int64_t>(erase_size);
1589 : :
1590 : 8 : auto& url_info = url_iter->second;
1591 : 8 : auto& total_url_info = total_url_iter->second;
1592 : 8 : --url_info.count;
1593 : 8 : url_info.size -= erase_size;
1594 : 8 : --total_url_info.count;
1595 : 8 : total_url_info.size -= erase_size;
1596 : :
1597 [ + + ]: 8 : if (url_info.count == 0) {
1598 : 6 : url_info.first_timestamp = -1;
1599 : 6 : url_info.last_timestamp = -1;
1600 : 6 : url_info.freq = 0;
1601 [ + - ]: 2 : } else if (url_info.first_timestamp >= 0) {
1602 [ - + ]: 2 : if (url_info.count == 1) {
1603 : 0 : url_info.first_timestamp = url_info.last_timestamp;
1604 : 0 : url_info.freq = 0;
1605 : : } else {
1606 : 2 : ::sqlite3_stmt* timestamp_stmt = nullptr;
1607 [ + - ]: 2 : ret = ::sqlite3_prepare_v2(
1608 : 2 : impl_->db, "SELECT elapsed FROM VLinkDatas WHERE rowid > ? AND url = ? ORDER BY rowid LIMIT 1;", -1,
1609 : : ×tamp_stmt, nullptr);
1610 : :
1611 [ + - ]: 2 : if VLIKELY (ret == SQLITE_OK) {
1612 [ + - ]: 2 : ::sqlite3_bind_int64(timestamp_stmt, get_column(0), erase_row_id);
1613 [ + - ]: 2 : ::sqlite3_bind_int(timestamp_stmt, get_column(1), erase_url_id);
1614 [ + - ]: 2 : ret = ::sqlite3_step(timestamp_stmt);
1615 : : }
1616 : :
1617 [ - + ]: 2 : if VUNLIKELY (ret != SQLITE_ROW) {
1618 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to refresh URL metadata after eviction: %s.", ::sqlite3_errmsg(impl_->db));
# # ]
1619 [ # # ]: 0 : if (timestamp_stmt) {
1620 [ # # ]: 0 : ::sqlite3_finalize(timestamp_stmt);
1621 : : }
1622 [ # # ]: 0 : ::sqlite3_finalize(delete_row_stmt);
1623 [ # # ]: 0 : rollback_cache();
1624 : 0 : return false;
1625 : : }
1626 : :
1627 [ + - ]: 2 : url_info.first_timestamp = ::sqlite3_column_int64(timestamp_stmt, 0);
1628 [ + - ]: 2 : ::sqlite3_finalize(timestamp_stmt);
1629 : :
1630 : 2 : const auto duration = (url_info.last_timestamp - url_info.first_timestamp) / 1000'000.0;
1631 [ + - ]: 2 : url_info.freq = duration > 0 ? url_info.count / duration : 0;
1632 : : }
1633 : : }
1634 : :
1635 [ + + ]: 8 : if (total_url_info.count == 0) {
1636 : 6 : total_url_info.first_timestamp = -1;
1637 : 6 : total_url_info.last_timestamp = -1;
1638 : 6 : total_url_info.freq = 0;
1639 [ + - ]: 2 : } else if (total_url_info.count == url_info.count) {
1640 : 2 : total_url_info.first_timestamp = url_info.first_timestamp;
1641 : 2 : total_url_info.last_timestamp = url_info.last_timestamp;
1642 : 2 : total_url_info.freq = url_info.freq;
1643 [ # # ]: 0 : } else if (total_url_info.first_timestamp >= 0) {
1644 [ # # ]: 0 : if (url_info.count > 0) {
1645 : 0 : total_url_info.last_timestamp = url_info.last_timestamp;
1646 : : } else {
1647 : 0 : total_url_info.last_timestamp = total_url_info.previous_segment_last_timestamp;
1648 : : }
1649 : :
1650 : 0 : const auto duration = (total_url_info.last_timestamp - total_url_info.first_timestamp) / 1000'000.0;
1651 [ # # ]: 0 : total_url_info.freq = duration > 0 ? total_url_info.count / duration : 0;
1652 : : }
1653 : : } else {
1654 : : // LCOV_EXCL_START GCOVR_EXCL_START
1655 : : CLOG_W("Failed to erase datas table: %s.", ::sqlite3_errmsg(impl_->db));
1656 : : if VLIKELY (delete_row_stmt) {
1657 : : ::sqlite3_finalize(delete_row_stmt);
1658 : : delete_row_stmt = nullptr;
1659 : : }
1660 : : if VLIKELY (select_stmt) {
1661 : : ::sqlite3_finalize(select_stmt);
1662 : : select_stmt = nullptr;
1663 : : }
1664 : : rollback_cache();
1665 : : return false;
1666 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1667 : : }
1668 : : } else {
1669 [ # # ]: 0 : if (ret == SQLITE_DONE) {
1670 [ # # # # ]: 0 : CLOG_W("VDBWriter: No data available while enforcing the configured limit.");
1671 : : } else {
1672 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to select data while enforcing the configured limit: %s.",
# # ]
1673 : : ::sqlite3_errmsg(impl_->db));
1674 : : }
1675 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1676 [ # # ]: 0 : rollback_cache();
1677 : 0 : return false;
1678 : : }
1679 : :
1680 [ + - ]: 8 : if VLIKELY (delete_row_stmt) {
1681 [ + - ]: 8 : ::sqlite3_finalize(delete_row_stmt);
1682 : 8 : delete_row_stmt = nullptr;
1683 : : }
1684 : :
1685 [ - + ]: 8 : if VLIKELY (select_stmt) {
1686 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1687 : 0 : select_stmt = nullptr;
1688 : : }
1689 : : } else {
1690 : : CLOG_W("Failed to erase datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1691 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1692 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1693 : : }
1694 : : }
1695 : :
1696 : : // split
1697 : :
1698 [ + + + + : 345 : if (impl_->is_split_mode.load(std::memory_order_acquire) && !impl_->url_map.empty()) {
+ + ]
1699 [ + + + + ]: 38 : if (impl_->config.split_by_time > 0 &&
1700 : 4 : (microseconds_timestamp - impl_->config.begin_time * 1000) >
1701 [ + - ]: 4 : impl_->config.split_by_time * 1000 * static_cast<int64_t>(impl_->split_file_list.size())) {
1702 : 4 : do_split = true;
1703 [ + - + - : 60 : } else if (impl_->config.split_by_time <= 0 && impl_->config.split_by_size > 0 &&
+ + ]
1704 [ + + ]: 30 : (impl_->current_size + static_cast<int64_t>(data.size())) > impl_->config.split_by_size) {
1705 : 25 : do_split = true;
1706 : : } else {
1707 : : do_split = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1708 : : }
1709 : :
1710 [ + + ]: 34 : if VUNLIKELY (do_split) {
1711 [ + - ]: 29 : std::lock_guard split_lock(impl_->split_mtx);
1712 : :
1713 : 29 : impl_->split_index.fetch_add(1, std::memory_order_relaxed);
1714 [ + - ]: 29 : impl_->time_current = impl_->time_start + std::chrono::milliseconds(microseconds_timestamp / 1000U);
1715 : :
1716 [ + + ]: 29 : if (impl_->config.split_name_by_time) {
1717 [ + + ]: 4 : if (impl_->base_dir.empty()) {
1718 [ + - + - ]: 2 : impl_->split_filename = get_format_date(&impl_->time_current, true) + ".vdb";
1719 : : } else {
1720 [ + - + - : 2 : impl_->split_filename = impl_->base_dir + "/" + get_format_date(&impl_->time_current, true) + ".vdb";
+ - + - ]
1721 : : }
1722 : : } else {
1723 : 25 : impl_->split_filename =
1724 [ + - + - : 75 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vdb";
+ - + - ]
1725 : : }
1726 : :
1727 [ + + + - : 29 : if (impl_->split_before && impl_->split_callback) {
+ + ]
1728 [ + - ]: 46 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
1729 : : }
1730 : :
1731 [ + - - + ]: 29 : if VUNLIKELY (!sync_cache()) {
1732 : : impl_->split_index.fetch_sub(1, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1733 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1734 : : }
1735 : :
1736 [ + - ]: 29 : close_segment();
1737 : :
1738 [ + - - + ]: 29 : if VUNLIKELY (!write_filex(false)) {
1739 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1740 : : }
1741 : :
1742 [ + - ]: 29 : open_split(impl_->split_filename);
1743 : :
1744 [ + - - + ]: 29 : if VUNLIKELY (!begin_cache()) {
1745 : : impl_->split_index.fetch_sub(1, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1746 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1747 : : }
1748 : :
1749 [ + + + + : 29 : if (!impl_->split_before && impl_->split_callback) {
+ + ]
1750 [ + - ]: 6 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
1751 : : }
1752 [ + - ]: 29 : }
1753 : : }
1754 : :
1755 [ + - ]: 345 : auto total_url_iter_ret = impl_->total_url_map.try_emplace(url, Impl::UrlMsgInfo());
1756 [ + - ]: 345 : auto url_iter_ret = impl_->url_map.try_emplace(url, Impl::UrlMsgInfo());
1757 : :
1758 : 14 : auto discard_new_url_entries = [this, &url_iter_ret, &total_url_iter_ret]() {
1759 [ + + ]: 5 : if (url_iter_ret.second) {
1760 : 1 : impl_->url_map.erase(url_iter_ret.first);
1761 : : }
1762 : :
1763 [ + + ]: 5 : if (total_url_iter_ret.second) {
1764 : 1 : impl_->total_url_map.erase(total_url_iter_ret.first);
1765 : : }
1766 : 350 : };
1767 : :
1768 : 345 : Impl::UrlMsgInfo& total_url_msg_info = total_url_iter_ret.first->second;
1769 : 345 : Impl::UrlMsgInfo& url_msg_info = url_iter_ret.first->second;
1770 : 345 : auto resolved_schema_type = SchemaData::resolve_type(schema_type, ser_type);
1771 : :
1772 [ + - ]: 345 : std::string next_ser_type = total_url_msg_info.ser_type;
1773 : 345 : SchemaType next_schema_type = total_url_msg_info.schema_type;
1774 : :
1775 [ + + ]: 345 : if (total_url_iter_ret.second) {
1776 [ + - ]: 181 : next_ser_type = ser_type;
1777 : 181 : next_schema_type = resolved_schema_type;
1778 : : } else {
1779 [ + + ]: 164 : if (!ser_type.empty()) {
1780 [ + + ]: 71 : if (next_ser_type.empty()) {
1781 [ + - ]: 1 : next_ser_type = ser_type;
1782 [ + + ]: 70 : } else if VUNLIKELY (next_ser_type != ser_type) {
1783 [ + - + - ]: 6 : CLOG_E("VDBWriter: URL [%s] ser changed from [%s] to [%s].", url.c_str(), next_ser_type.c_str(),
1784 : : ser_type.c_str());
1785 [ + - ]: 3 : discard_new_url_entries();
1786 : 3 : return false;
1787 : : }
1788 : : }
1789 : : }
1790 : :
1791 [ + + ]: 342 : if (!next_ser_type.empty()) {
1792 : 242 : std::string schema_ser_type;
1793 [ - + ]: 242 : const auto schema_ser_source = ser_type.empty() ? std::string_view{next_ser_type} : std::string_view{ser_type};
1794 : 242 : SchemaType schema_storage_type = SchemaData::resolve_type(schema_type, schema_ser_source);
1795 : 242 : bool has_split_method_schema = false;
1796 : :
1797 [ + - ]: 242 : schema_ser_type.assign(schema_ser_source.begin(), schema_ser_source.end());
1798 : :
1799 [ + + + + ]: 210 : if ((action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
1800 [ + + + + : 485 : action_type == ActionType::kServerRequest || action_type == ActionType::kServerResponse) &&
+ + ]
1801 [ + - ]: 36 : !schema_ser_source.empty()) {
1802 : 36 : const auto split_pos = schema_ser_source.find('|');
1803 : :
1804 [ + - ]: 36 : if (split_pos != std::string_view::npos) {
1805 [ + - ]: 36 : auto payload_ser_type = schema_ser_source.substr(0, split_pos);
1806 : :
1807 [ + + + + ]: 36 : if (action_type == ActionType::kClientResponse || action_type == ActionType::kServerResponse) {
1808 [ + - ]: 3 : payload_ser_type = schema_ser_source.substr(split_pos + 1);
1809 : : }
1810 : :
1811 [ + - ]: 36 : if (!payload_ser_type.empty()) {
1812 [ + - ]: 36 : schema_ser_type.assign(payload_ser_type.begin(), payload_ser_type.end());
1813 : 36 : schema_storage_type = SchemaData::resolve_type(schema_type, payload_ser_type);
1814 : 36 : has_split_method_schema = true;
1815 : : }
1816 : : }
1817 : : }
1818 : 242 : SchemaData schema_data;
1819 : :
1820 [ + - + + ]: 242 : if VUNLIKELY (!load_schema(schema_ser_type, schema_storage_type, schema_data)) {
1821 [ + - ]: 2 : discard_new_url_entries();
1822 : 2 : return false;
1823 : : }
1824 : :
1825 : 240 : schema_storage_type = SchemaData::resolve_type(schema_storage_type, schema_ser_type, schema_data.encoding);
1826 : :
1827 [ + + ]: 240 : if (has_split_method_schema) {
1828 [ + + ]: 36 : if (schema_storage_type != SchemaType::kUnknown) {
1829 [ - + ]: 6 : if (next_schema_type == SchemaType::kUnknown) {
1830 : : next_schema_type = schema_storage_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1831 [ - + ]: 6 : } else if (next_schema_type != schema_storage_type) {
1832 : : next_schema_type = SchemaType::kUnknown; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1833 : : }
1834 : : }
1835 : : } else {
1836 [ + + ]: 204 : if (resolved_schema_type == SchemaType::kUnknown) {
1837 : : const auto inferred_schema_type =
1838 : 1 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
1839 : :
1840 [ + - ]: 1 : if (inferred_schema_type != SchemaType::kUnknown) {
1841 : 1 : resolved_schema_type = inferred_schema_type;
1842 : : }
1843 : : }
1844 : :
1845 [ + - ]: 204 : if (resolved_schema_type != SchemaType::kUnknown) {
1846 [ + + ]: 204 : if (next_schema_type == SchemaType::kUnknown) {
1847 : 2 : next_schema_type = resolved_schema_type;
1848 [ - + ]: 202 : } else if VUNLIKELY (next_schema_type != resolved_schema_type) {
1849 : : // LCOV_EXCL_START GCOVR_EXCL_START
1850 : : CLOG_E("VDBWriter: URL [%s] schema changed from [%d] to [%d].", url.c_str(),
1851 : : static_cast<int>(next_schema_type), static_cast<int>(resolved_schema_type));
1852 : : discard_new_url_entries();
1853 : : return false;
1854 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1855 : : }
1856 : : }
1857 : : }
1858 : :
1859 [ + - ]: 240 : std::string schema_key = schema_ser_type;
1860 [ + - ]: 240 : schema_key.push_back('\x1F');
1861 [ + - ]: 240 : schema_key.append(SchemaData::convert_type(schema_storage_type));
1862 : :
1863 [ + - + + ]: 240 : if (impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
1864 : : const bool has_schema_blob =
1865 [ + + + - : 156 : !schema_data.name.empty() && !schema_data.encoding.empty() && !schema_data.data.empty();
+ - ]
1866 : :
1867 [ + + ]: 156 : if (has_schema_blob) {
1868 [ + - - + ]: 8 : if VUNLIKELY (!insert_schema(schema_data)) {
1869 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1870 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1871 : : }
1872 : : }
1873 : :
1874 [ + - ]: 156 : impl_->ser_map.emplace(schema_key);
1875 : : }
1876 [ + - + + : 244 : }
+ + ]
1877 : :
1878 [ + + ]: 340 : if (total_url_iter_ret.second) {
1879 : 180 : total_url_msg_info.index = impl_->total_url_map.size() - 1;
1880 [ + - ]: 180 : impl_->total_url_list.emplace_back(url);
1881 : : }
1882 : :
1883 [ + + ]: 340 : if (url_iter_ret.second) {
1884 : : // insert url
1885 : 208 : url_msg_info.index = impl_->url_map.size() - 1;
1886 : :
1887 [ + + ]: 208 : if (!total_url_iter_ret.second) {
1888 : 28 : total_url_msg_info.previous_segment_last_timestamp = total_url_msg_info.last_timestamp;
1889 : : }
1890 : :
1891 [ + + + - : 208 : if (action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
+ - ]
1892 [ - + ]: 176 : action_type == ActionType::kServerRequest ||
1893 : : action_type == ActionType::kServerResponse) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1894 [ + - ]: 32 : impl_->write_url_type = "Method";
1895 [ + + + + ]: 176 : } else if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe) {
1896 [ + - ]: 141 : impl_->write_url_type = "Event";
1897 [ + + + + ]: 35 : } else if (action_type == ActionType::kSet || action_type == ActionType::kGet) {
1898 [ + - ]: 34 : impl_->write_url_type = "Field";
1899 : : } else {
1900 [ + - ]: 1 : impl_->write_url_type = "Unknown";
1901 : : }
1902 : :
1903 [ + - ]: 208 : ::sqlite3_bind_int(impl_->urls_stmt, get_column(0), total_url_msg_info.index);
1904 [ + - ]: 208 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(1), url.c_str(), url.size(), SQLITE_STATIC);
1905 [ + - ]: 208 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(2), impl_->write_url_type.c_str(), impl_->write_url_type.size(),
1906 : : SQLITE_STATIC);
1907 [ + - ]: 208 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(3), next_ser_type.c_str(), next_ser_type.size(), SQLITE_STATIC);
1908 [ + - ]: 208 : const std::string next_encoding(SchemaData::convert_type(next_schema_type));
1909 [ + - ]: 208 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(4), next_encoding.c_str(), next_encoding.size(), SQLITE_STATIC);
1910 [ + - ]: 208 : ::sqlite3_bind_int64(impl_->urls_stmt, get_column(5), 0);
1911 [ + - ]: 208 : ::sqlite3_bind_double(impl_->urls_stmt, get_column(6), 0);
1912 [ + - ]: 208 : ::sqlite3_bind_int64(impl_->urls_stmt, get_column(7), 0);
1913 [ + - ]: 208 : ::sqlite3_bind_double(impl_->urls_stmt, get_column(8), 0);
1914 : :
1915 [ + - ]: 208 : ret = ::sqlite3_step(impl_->urls_stmt);
1916 : :
1917 [ - + ]: 208 : if VUNLIKELY (ret != SQLITE_DONE) {
1918 : : CLOG_W("Failed to insert urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1919 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1920 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1921 : : }
1922 : :
1923 [ + - ]: 208 : ret = ::sqlite3_reset(impl_->urls_stmt);
1924 : :
1925 [ - + ]: 208 : if VUNLIKELY (ret != SQLITE_OK) {
1926 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1927 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1928 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1929 : : }
1930 : :
1931 [ + - ]: 208 : url_msg_info.url_type = impl_->write_url_type;
1932 [ + - ]: 208 : total_url_msg_info.url_type = impl_->write_url_type;
1933 : :
1934 [ + + ]: 208 : if (!next_ser_type.empty()) {
1935 [ + - ]: 201 : url_msg_info.ser_type = next_ser_type;
1936 [ + - ]: 201 : total_url_msg_info.ser_type = next_ser_type;
1937 : : }
1938 : :
1939 : 208 : url_msg_info.schema_type = next_schema_type;
1940 : 208 : total_url_msg_info.schema_type = next_schema_type;
1941 [ + - + + : 340 : } else if (total_url_msg_info.ser_type != next_ser_type || total_url_msg_info.schema_type != next_schema_type) {
- + + + ]
1942 : 1 : const bool ser_changed = total_url_msg_info.ser_type != next_ser_type;
1943 : 1 : const bool schema_changed = total_url_msg_info.schema_type != next_schema_type;
1944 : :
1945 [ - + - - ]: 1 : if (ser_changed || schema_changed) {
1946 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(0), next_ser_type.c_str(), next_ser_type.size(),
1947 : : SQLITE_STATIC);
1948 [ + - ]: 1 : const std::string next_encoding(SchemaData::convert_type(next_schema_type));
1949 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(1), next_encoding.c_str(), next_encoding.size(),
1950 : : SQLITE_STATIC);
1951 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(2), url.c_str(), url.size(), SQLITE_STATIC);
1952 : :
1953 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_url_meta_stmt);
1954 : :
1955 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
1956 : : // LCOV_EXCL_START GCOVR_EXCL_START
1957 : : CLOG_W("Failed to update urls metadata table: %s.", ::sqlite3_errmsg(impl_->db));
1958 : : rollback_cache();
1959 : : return false;
1960 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1961 : : }
1962 : :
1963 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_url_meta_stmt);
1964 : :
1965 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
1966 : : // LCOV_EXCL_START GCOVR_EXCL_START
1967 : : CLOG_W("Failed to reset urls metadata table: %s.", ::sqlite3_errmsg(impl_->db));
1968 : : rollback_cache();
1969 : : return false;
1970 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1971 : : }
1972 : :
1973 [ + - ]: 1 : total_url_msg_info.ser_type = next_ser_type;
1974 [ + - ]: 1 : url_msg_info.ser_type = next_ser_type;
1975 : 1 : total_url_msg_info.schema_type = next_schema_type;
1976 : 1 : url_msg_info.schema_type = next_schema_type;
1977 [ + - ]: 1 : }
1978 : : }
1979 : :
1980 : : // update info
1981 : 340 : ++url_msg_info.count;
1982 : 340 : ++total_url_msg_info.count;
1983 : :
1984 : 340 : impl_->cached_size.fetch_add(data.size(), std::memory_order_relaxed);
1985 : :
1986 : 340 : ++impl_->current_row;
1987 : 340 : ++impl_->total_current_row;
1988 : 340 : impl_->total_timestamp = std::max(impl_->total_timestamp, microseconds_timestamp);
1989 : :
1990 [ + + + + : 340 : if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe || action_type == ActionType::kSet ||
+ + + + ]
1991 : : action_type == ActionType::kGet) {
1992 : 303 : double time_duration = 0;
1993 : :
1994 [ + + ]: 303 : if (total_url_msg_info.first_timestamp < 0) {
1995 : 153 : total_url_msg_info.first_timestamp = microseconds_timestamp;
1996 : : }
1997 : :
1998 : 303 : total_url_msg_info.last_timestamp = microseconds_timestamp;
1999 : 303 : time_duration = (total_url_msg_info.last_timestamp - total_url_msg_info.first_timestamp) / 1000'000.0;
2000 : :
2001 [ + + ]: 303 : if (time_duration > 0) {
2002 : 149 : total_url_msg_info.freq = total_url_msg_info.count / time_duration;
2003 : : } else {
2004 : 154 : total_url_msg_info.freq = 0;
2005 : : }
2006 : :
2007 [ + + ]: 303 : if (url_msg_info.first_timestamp < 0) {
2008 : 181 : url_msg_info.first_timestamp = microseconds_timestamp;
2009 : : }
2010 : :
2011 : 303 : url_msg_info.last_timestamp = microseconds_timestamp;
2012 : 303 : time_duration = (url_msg_info.last_timestamp - url_msg_info.first_timestamp) / 1000'000.0;
2013 : :
2014 [ + + ]: 303 : if (time_duration > 0) {
2015 : 121 : url_msg_info.freq = url_msg_info.count / time_duration;
2016 : : } else {
2017 : 182 : url_msg_info.freq = 0;
2018 : : }
2019 : : }
2020 : :
2021 : : // insert data
2022 [ + - ]: 340 : const auto write_action_type = convert_action(action_type);
2023 : :
2024 : 340 : impl_->last_timestamp = std::max(impl_->last_timestamp, microseconds_timestamp);
2025 : :
2026 [ + - ]: 340 : ::sqlite3_bind_int64(impl_->datas_stmt, get_column(0), microseconds_timestamp);
2027 [ + - ]: 340 : ::sqlite3_bind_int(impl_->datas_stmt, get_column(1), total_url_msg_info.index);
2028 [ + - ]: 340 : ::sqlite3_bind_text(impl_->datas_stmt, get_column(2), write_action_type.data(), write_action_type.size(),
2029 : : SQLITE_STATIC);
2030 : :
2031 : : // check compress
2032 : 340 : do_compress = false;
2033 : :
2034 [ + + + - : 367 : if (impl_->enable_compressed && static_cast<int64_t>(data.size()) >= impl_->config.compress_start_size &&
+ + ]
2035 [ + - + + ]: 27 : impl_->config.ignore_compress_urls.count(url) == 0) {
2036 [ + - ]: 18 : auto& compress_ignore_cnt = impl_->compress_ignore_map[url];
2037 : :
2038 [ + + ]: 18 : if (compress_ignore_cnt < kCompressMaxIgnoreCnt) {
2039 : 17 : compressed_data = Bytes::compress_data(data.data(), data.size(), impl_->config.compress_level > 3);
2040 : :
2041 [ + - + + : 17 : if (!compressed_data.empty() && compressed_data.size() < data.size() * kCompressMaxRatio) {
+ + ]
2042 : 12 : do_compress = true;
2043 : 12 : compress_ignore_cnt = 0;
2044 : : } else {
2045 : 5 : ++compress_ignore_cnt;
2046 : : }
2047 : : }
2048 : : }
2049 : :
2050 [ + + ]: 340 : if (do_compress) {
2051 [ + - ]: 12 : ::sqlite3_bind_blob(impl_->datas_stmt, get_column(3), compressed_data.data(), compressed_data.size(),
2052 : : SQLITE_STATIC);
2053 : : } else {
2054 [ + - ]: 328 : ::sqlite3_bind_blob(impl_->datas_stmt, get_column(3), data.data(), data.size(), SQLITE_STATIC);
2055 : : }
2056 : :
2057 : 340 : auto accounted_size = data.size();
2058 : :
2059 [ + + + + : 340 : if VUNLIKELY (!do_compress && impl_->enable_compressed && Bytes::is_compress_data(data.data(), data.size())) {
+ + + + +
+ ]
2060 : 6 : accounted_size = (static_cast<size_t>(data[4]) << 24U) | (static_cast<size_t>(data[5]) << 16U) |
2061 : 6 : (static_cast<size_t>(data[6]) << 8U) | static_cast<size_t>(data[7]);
2062 : : }
2063 : :
2064 : 340 : url_msg_info.size += accounted_size;
2065 : 340 : total_url_msg_info.size += accounted_size;
2066 : 340 : impl_->current_size += static_cast<int64_t>(accounted_size);
2067 : 340 : impl_->total_current_size += static_cast<int64_t>(accounted_size);
2068 : :
2069 [ + - ]: 340 : ret = ::sqlite3_step(impl_->datas_stmt);
2070 : :
2071 [ - + ]: 340 : if VUNLIKELY (ret != SQLITE_DONE) {
2072 : : CLOG_W("Failed to insert datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2073 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2074 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2075 : : }
2076 : :
2077 [ + - ]: 340 : ret = ::sqlite3_reset(impl_->datas_stmt);
2078 : :
2079 [ - + ]: 340 : if VUNLIKELY (ret != SQLITE_OK) {
2080 : : CLOG_W("Failed to reset datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2081 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2082 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2083 : : }
2084 : :
2085 : : // update on wal_mode
2086 : :
2087 [ + + ]: 340 : if (impl_->sync_timer.get() > kSyncWriteInterval) {
2088 : 1 : impl_->sync_timer.restart();
2089 : :
2090 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(0), url_msg_info.count);
2091 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(1), url_msg_info.size);
2092 [ + - ]: 1 : ::sqlite3_bind_double(impl_->update_urls_stmt, get_column(2), url_msg_info.freq);
2093 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_urls_stmt, get_column(3), url.c_str(), url.size(), SQLITE_STATIC);
2094 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_urls_stmt);
2095 : :
2096 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
2097 : : CLOG_W("Failed to update urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2098 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2099 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2100 : : }
2101 : :
2102 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_urls_stmt);
2103 : :
2104 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
2105 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2106 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2107 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2108 : : }
2109 : :
2110 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(0), impl_->current_row);
2111 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(1), impl_->last_timestamp);
2112 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_header_stmt);
2113 : :
2114 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
2115 : : CLOG_W("Failed to update header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2116 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2117 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2118 : : }
2119 : :
2120 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_header_stmt);
2121 : :
2122 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
2123 : : CLOG_W("Failed to reset header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2124 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2125 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2126 : : }
2127 : : }
2128 : :
2129 [ + + ]: 680 : if VUNLIKELY (impl_->cached_size.load(std::memory_order_relaxed) > impl_->config.cache_size) {
2130 [ + - - + ]: 46 : if VUNLIKELY (!sync_cache()) {
2131 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2132 : : }
2133 : : }
2134 : :
2135 : 340 : return true;
2136 : : #else
2137 : : (void)url;
2138 : : (void)ser_type;
2139 : : (void)schema_type;
2140 : : (void)action_type;
2141 : : (void)data;
2142 : : (void)microseconds_timestamp;
2143 : : return false;
2144 : : #endif
2145 : 347 : }
2146 : :
2147 : 98 : bool VDBWriter::write_filex(bool complete) {
2148 : : #ifdef _WIN32
2149 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
2150 : : #else
2151 [ + - ]: 98 : std::filesystem::path file_path(impl_->path);
2152 : : #endif
2153 : :
2154 : : try {
2155 : 98 : nlohmann::ordered_json json;
2156 : :
2157 [ + - ]: 98 : json["VLinkHeader"] = {
2158 : : {"major", VLINK_VERSION_MAJOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2159 : : {"minor", VLINK_VERSION_MINOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2160 : : {"patch", VLINK_VERSION_PATCH}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2161 : 98 : {"count", impl_->total_current_row},
2162 : 98 : {"duration", impl_->total_timestamp},
2163 : : {"accuracy", "MicroSecond"},
2164 : 98 : {"compress", impl_->enable_compressed ? "lzav" : "None"},
2165 : 98 : {"process", impl_->app_name},
2166 [ + - ]: 98 : {"date", get_format_date(&impl_->time_start)},
2167 : 98 : {"tag", impl_->tag_name},
2168 [ + + ]: 98 : {"split_by_size", impl_->config.split_by_time > 0 ? 0 : impl_->config.split_by_size},
2169 : 98 : {"split_by_time", impl_->config.split_by_time},
2170 : : {"complete", complete},
2171 : 98 : {"timezone", impl_->timezone_diff},
2172 : 98 : {"start_timestamp", impl_->start_timestamp},
2173 [ + - + - : 5586 : };
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
+ + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - ]
2174 : :
2175 : 98 : nlohmann::ordered_json url_json;
2176 : :
2177 : : {
2178 [ + - + - ]: 98 : std::lock_guard lock(sample_mutex());
2179 : :
2180 [ + + ]: 164 : for (const auto& url : impl_->total_url_list) {
2181 [ + - ]: 66 : const auto& ext_info = impl_->total_url_map[url];
2182 [ + - + - : 66 : auto loss = total_url_loss_map_ref()[recover_recorded_url(url)];
+ - ]
2183 : :
2184 [ + - + - : 2376 : url_json.push_back({
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + +
+ + + + +
+ + + + +
+ + + + +
+ + + - -
- - - - -
- - - - -
- - - - -
- - - ]
2185 : : // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2186 : 66 : {"index", ext_info.index},
2187 : : {"url", url},
2188 [ + - ]: 66 : {"type", ext_info.url_type},
2189 [ + - ]: 66 : {"ser", ext_info.ser_type},
2190 [ + - ]: 132 : {"encoding", std::string(SchemaData::convert_type(ext_info.schema_type))},
2191 : 66 : {"count", ext_info.count},
2192 : 66 : {"size", ext_info.size},
2193 : : {"loss", loss},
2194 : 66 : {"freq", ext_info.freq},
2195 : : });
2196 : : }
2197 : 98 : }
2198 : :
2199 [ + - ]: 98 : json["VLinkUrls"] = std::move(url_json);
2200 : :
2201 : 98 : nlohmann::ordered_json files_json;
2202 [ + + ]: 195 : for (const auto& file : impl_->split_file_list) {
2203 [ + - + - ]: 97 : files_json.push_back(file);
2204 : : }
2205 : :
2206 [ + - ]: 98 : json["VLinkFiles"] = std::move(files_json);
2207 : :
2208 [ + - ]: 98 : std::ofstream filex(file_path);
2209 : :
2210 [ + - - + ]: 98 : if VUNLIKELY (!filex.is_open()) {
2211 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2212 : : }
2213 : :
2214 [ + - + - ]: 98 : filex << json.dump(4);
2215 [ + - ]: 98 : filex.close();
2216 : :
2217 [ + - - + ]: 98 : if VUNLIKELY (!filex) {
2218 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2219 : : }
2220 [ + - + - : 98 : } catch (const nlohmann::json::exception& e) {
+ - + - -
- ]
2221 : : CLOG_W("VDBWriter: JSON error during config export: %s.", e.what()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2222 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2223 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2224 : :
2225 : 98 : return true;
2226 : 98 : }
2227 : :
2228 : 678 : bool VDBWriter::begin_cache() {
2229 : : #ifdef VLINK_ENABLE_SQLITE
2230 : :
2231 [ - + ]: 678 : if VUNLIKELY (!impl_->db) {
2232 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2233 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2234 : : }
2235 : :
2236 [ + + ]: 678 : if VLIKELY (impl_->in_cached.load(std::memory_order_relaxed)) {
2237 : 213 : return true;
2238 : : }
2239 : :
2240 : 465 : int ret = ::sqlite3_exec(impl_->db, "BEGIN;", nullptr, nullptr, nullptr);
2241 : :
2242 [ - + ]: 465 : if VUNLIKELY (ret != SQLITE_OK) {
2243 : : CLOG_W("Failed to begin transaction: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2244 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2245 : : }
2246 : :
2247 : 465 : impl_->cache_snapshot = std::make_unique<Impl::WriteStateSnapshot>(*impl_);
2248 : 465 : impl_->in_cached.store(true, std::memory_order_relaxed);
2249 : 465 : impl_->cached_size.store(0, std::memory_order_relaxed);
2250 : :
2251 : 465 : impl_->cache_timer.restart();
2252 : :
2253 : 465 : return true;
2254 : : #else
2255 : : return false;
2256 : : #endif
2257 : : }
2258 : :
2259 : 528 : bool VDBWriter::sync_cache() {
2260 : : #ifdef VLINK_ENABLE_SQLITE
2261 : :
2262 [ - + ]: 528 : if VUNLIKELY (!impl_->db) {
2263 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2264 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2265 : : }
2266 : :
2267 [ + + ]: 528 : if (!impl_->in_cached.load(std::memory_order_relaxed)) {
2268 : 65 : return true;
2269 : : }
2270 : :
2271 : 463 : int ret = ::sqlite3_exec(impl_->db, "COMMIT;", nullptr, nullptr, nullptr);
2272 : :
2273 [ - + ]: 463 : if VUNLIKELY (ret != SQLITE_OK) {
2274 : : CLOG_W("Failed to commit transaction: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2275 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2276 : : }
2277 : :
2278 : 463 : impl_->in_cached.store(false, std::memory_order_relaxed);
2279 : 463 : impl_->cached_size.store(0, std::memory_order_relaxed);
2280 : 463 : impl_->cache_snapshot.reset();
2281 : :
2282 : 463 : impl_->cache_timer.stop();
2283 : :
2284 : 463 : return true;
2285 : : #else
2286 : : return false;
2287 : : #endif
2288 : : }
2289 : :
2290 : 2 : bool VDBWriter::rollback_cache() {
2291 : : #ifdef VLINK_ENABLE_SQLITE
2292 : :
2293 [ - + ]: 2 : if VUNLIKELY (!impl_->db) {
2294 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2295 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2296 : : }
2297 : :
2298 : 2 : ::sqlite3_exec(impl_->db, "ROLLBACK;", nullptr, nullptr, nullptr);
2299 : :
2300 [ + - ]: 2 : if (impl_->cache_snapshot) {
2301 : 2 : impl_->cache_snapshot->restore(*impl_);
2302 : 2 : impl_->cache_snapshot.reset();
2303 : : }
2304 : :
2305 : 2 : impl_->in_cached.store(false, std::memory_order_relaxed);
2306 : 2 : impl_->cached_size.store(0, std::memory_order_relaxed);
2307 : :
2308 : 2 : impl_->cache_timer.stop();
2309 : :
2310 : 2 : return true;
2311 : : #else
2312 : : return false;
2313 : : #endif
2314 : : }
2315 : :
2316 : 44 : bool VDBWriter::insert_schema(const SchemaData& schema_data) {
2317 : : #ifdef VLINK_ENABLE_SQLITE
2318 : :
2319 [ - + ]: 44 : if VUNLIKELY (!impl_->db) {
2320 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2321 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2322 : : }
2323 : :
2324 [ + - ]: 44 : std::string encoding = schema_data.encoding;
2325 : 44 : SchemaType schema_type = SchemaData::resolve_type(schema_data.schema_type, schema_data.name, encoding);
2326 : :
2327 [ - + - - : 44 : if (encoding.empty() && SchemaData::is_real_type(schema_type)) {
- + ]
2328 : : encoding = std::string(SchemaData::convert_type(schema_type)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2329 : : }
2330 : :
2331 [ + - + - : 44 : if (schema_data.name.empty() || encoding.empty() || schema_data.data.empty()) {
- + - + ]
2332 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2333 : : }
2334 : :
2335 : 44 : int ret = 0;
2336 : :
2337 [ + - ]: 44 : ::sqlite3_bind_text(impl_->schemas_stmt, get_column(0), schema_data.name.c_str(), schema_data.name.size(),
2338 : : SQLITE_STATIC);
2339 : :
2340 [ + - ]: 44 : ::sqlite3_bind_text(impl_->schemas_stmt, get_column(1), encoding.c_str(), encoding.size(), SQLITE_STATIC);
2341 : :
2342 [ + - ]: 44 : ::sqlite3_bind_blob(impl_->schemas_stmt, get_column(2), schema_data.data.data(), schema_data.data.size(),
2343 : : SQLITE_STATIC);
2344 : :
2345 [ + - ]: 44 : ret = ::sqlite3_step(impl_->schemas_stmt);
2346 : :
2347 [ - + ]: 44 : if VUNLIKELY (ret != SQLITE_DONE) {
2348 : : CLOG_W("Failed to insert schema table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2349 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2350 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2351 : : }
2352 : :
2353 [ + - ]: 44 : ret = ::sqlite3_reset(impl_->schemas_stmt);
2354 : :
2355 [ - + ]: 44 : if VUNLIKELY (ret != SQLITE_OK) {
2356 : : CLOG_W("Failed to reset schema table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2357 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2358 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2359 : : }
2360 : :
2361 : 44 : return true;
2362 : : #else
2363 : : (void)schema_data;
2364 : : return false;
2365 : : #endif
2366 : 44 : }
2367 : :
2368 : : } // namespace vlink
|
