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