1 /*
2  * Copyright (c) 2023 Huawei Device Co., Ltd.
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15 #include "virtual_cloud_db.h"
16 
17 #include <thread>
18 #include "cloud/cloud_db_constant.h"
19 #include "cloud/cloud_db_types.h"
20 #include "db_constant.h"
21 #include "log_print.h"
22 #include "relational_store_manager.h"
23 #include "time_helper.h"
24 
25 namespace DistributedDB {
26 namespace {
27     const char *g_deleteField = CloudDbConstant::DELETE_FIELD;
28     const char *g_gidField = CloudDbConstant::GID_FIELD;
29     const char *g_cursorField = CloudDbConstant::CURSOR_FIELD;
30     const char *g_modifiedField = CloudDbConstant::MODIFY_FIELD;
31     const char *g_queryField = CloudDbConstant::QUERY_FIELD;
32 }
33 
BatchInsert(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend)34 DBStatus VirtualCloudDb::BatchInsert(const std::string &tableName, std::vector<VBucket> &&record,
35     std::vector<VBucket> &extend)
36 {
37     if (cloudError_) {
38         return DB_ERROR;
39     }
40     if (blockTimeMs_ != 0) {
41         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
42     }
43     if (record.size() != extend.size()) {
44         LOGE("[VirtualCloudDb] not equal records");
45         return DB_ERROR;
46     }
47     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
48     DBStatus status = InnerBatchInsert(tableName, std::move(record), extend);
49     if (status != OK) {
50         return status;
51     }
52     if (missingExtendCount_ > 0) {
53         extend.erase(extend.end());
54     } else if (missingExtendCount_ < 0) {
55         VBucket vBucket;
56         extend.push_back(vBucket);
57     }
58     if (insertFailedCount_ > 0) {
59         insertFailedCount_--;
60         LOGW("[VirtualCloud] Insert failed by testcase config");
61         return DB_ERROR;
62     }
63     if (cloudNetworkError_) {
64         return CLOUD_NETWORK_ERROR;
65     }
66     return OK;
67 }
68 
InnerBatchInsert(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend)69 DBStatus VirtualCloudDb::InnerBatchInsert(const std::string &tableName, std::vector<VBucket> &&record,
70     std::vector<VBucket> &extend)
71 {
72     DBStatus res = OK;
73     for (size_t i = 0; i < record.size(); ++i) {
74         if (insertCheckFunc_) {
75             insertCheckFunc_(record[i]);
76         }
77         if (extend[i].find(g_gidField) != extend[i].end()) {
78             LOGE("[VirtualCloudDb] Insert data should not have gid");
79             return DB_ERROR;
80         }
81         if (forkUploadConflictFunc_) {
82             DBStatus ret = forkUploadConflictFunc_(tableName, extend[i], record[i], cloudData_[tableName]);
83             if (ret != OK) {
84                 res = ret;
85                 continue;
86             }
87         }
88         if (conflictInUpload_) {
89             extend[i][CloudDbConstant::ERROR_FIELD] = static_cast<int64_t>(DBStatus::CLOUD_RECORD_EXIST_CONFLICT);
90         }
91         extend[i][g_gidField] = std::to_string(currentGid_++);
92         extend[i][g_cursorField] = std::to_string(currentCursor_++);
93         extend[i][g_deleteField] = false;
94         extend[i][CloudDbConstant::VERSION_FIELD] = std::to_string(currentVersion_++);
95         AddAssetIdForExtend(record[i], extend[i]);
96         if (forkUploadFunc_) {
97             forkUploadFunc_(tableName, extend[i]);
98         }
99         CloudData cloudData = {
100             .record = std::move(record[i]),
101             .extend = extend[i]
102         };
103         cloudData_[tableName].push_back(cloudData);
104         auto gid = std::get<std::string>(extend[i][g_gidField]);
105     }
106     return res;
107 }
108 
BatchInsertWithGid(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend)109 DBStatus VirtualCloudDb::BatchInsertWithGid(const std::string &tableName, std::vector<VBucket> &&record,
110     std::vector<VBucket> &extend)
111 {
112     if (cloudError_) {
113         return DB_ERROR;
114     }
115     if (blockTimeMs_ != 0) {
116         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
117     }
118     if (record.size() != extend.size()) {
119         LOGE("[VirtualCloudDb] not equal records");
120         return DB_ERROR;
121     }
122     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
123     for (size_t i = 0; i < record.size(); ++i) {
124         if (extend[i].find(g_gidField) == extend[i].end()) {
125             extend[i][g_gidField] = std::to_string(currentGid_++);
126         } else {
127             currentGid_++;
128         }
129         extend[i][g_cursorField] = std::to_string(currentCursor_++);
130         extend[i][g_deleteField] = false;
131         extend[i][CloudDbConstant::VERSION_FIELD] = std::to_string(currentVersion_++);
132         AddAssetIdForExtend(record[i], extend[i]);
133         if (forkUploadFunc_) {
134             forkUploadFunc_(tableName, extend[i]);
135         }
136         CloudData cloudData = {
137             .record = std::move(record[i]),
138             .extend = extend[i]
139         };
140         cloudData_[tableName].push_back(cloudData);
141     }
142     if (missingExtendCount_ > 0) {
143         extend.erase(extend.end());
144     } else if (missingExtendCount_ < 0) {
145         VBucket vBucket;
146         extend.push_back(vBucket);
147     }
148     if (cloudNetworkError_) {
149         return CLOUD_NETWORK_ERROR;
150     }
151     LOGI("[VirtualCloudDb] BatchInsertWithGid records");
152     return OK;
153 }
154 
BatchUpdate(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend)155 DBStatus VirtualCloudDb::BatchUpdate(const std::string &tableName, std::vector<VBucket> &&record,
156     std::vector<VBucket> &extend)
157 {
158     if (cloudError_) {
159         return DB_ERROR;
160     }
161     if (blockTimeMs_ != 0) {
162         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
163     }
164     return InnerUpdate(tableName, std::move(record), extend, false);
165 }
166 
BatchDelete(const std::string & tableName,std::vector<VBucket> & extend)167 DBStatus VirtualCloudDb::BatchDelete(const std::string &tableName, std::vector<VBucket> &extend)
168 {
169     if (cloudError_) {
170         return DB_ERROR;
171     }
172     if (blockTimeMs_ != 0) {
173         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
174     }
175     std::vector<VBucket> record;
176     for (size_t i = 0; i < extend.size(); ++i) {
177         record.emplace_back();
178     }
179     return InnerUpdate(tableName, std::move(record), extend, true);
180 }
181 
HeartBeat()182 DBStatus VirtualCloudDb::HeartBeat()
183 {
184     heartbeatCount_++;
185     if (actionStatus_ != OK) {
186         return actionStatus_;
187     }
188     if (cloudError_) {
189         return DB_ERROR;
190     }
191     if (heartbeatBlockTimeMs_ != 0) {
192         std::this_thread::sleep_for(std::chrono::milliseconds(heartbeatBlockTimeMs_));
193     }
194     if (heartbeatError_) {
195         return DB_ERROR;
196     }
197     if (blockTimeMs_ != 0) {
198         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
199     }
200     lockStatus_ = true;
201     return OK;
202 }
203 
Lock()204 std::pair<DBStatus, uint32_t> VirtualCloudDb::Lock()
205 {
206     if (lockStatus_) {
207         return { CLOUD_LOCK_ERROR, DBConstant::MIN_TIMEOUT };
208     }
209     lockCount_++;
210     if (actionStatus_ != OK) {
211         return { actionStatus_, DBConstant::MIN_TIMEOUT };
212     }
213     if (cloudError_) {
214         return { DB_ERROR, DBConstant::MIN_TIMEOUT };
215     }
216     if (blockTimeMs_ != 0) {
217         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
218     }
219     lockStatus_ = true;
220     return { OK, DBConstant::MIN_TIMEOUT };
221 }
222 
UnLock()223 DBStatus VirtualCloudDb::UnLock()
224 {
225     if (cloudError_) {
226         return DB_ERROR;
227     }
228     if (blockTimeMs_ != 0) {
229         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
230     }
231     lockStatus_ = false;
232     return OK;
233 }
234 
Close()235 DBStatus VirtualCloudDb::Close()
236 {
237     if (cloudError_) {
238         return DB_ERROR;
239     }
240     return OK;
241 }
242 
GetEmptyCursor(const std::string & tableName)243 std::pair<DBStatus, std::string> VirtualCloudDb::GetEmptyCursor(const std::string &tableName)
244 {
245     return { OK, "0" };
246 }
247 
DeleteByGid(const std::string & tableName,VBucket & extend)248 DBStatus VirtualCloudDb::DeleteByGid(const std::string &tableName, VBucket &extend)
249 {
250     for (auto &tableData : cloudData_[tableName]) {
251         if (std::get<std::string>(tableData.extend[g_gidField]) == std::get<std::string>(extend[g_gidField])) {
252             tableData.extend[g_modifiedField] = (int64_t)TimeHelper::GetSysCurrentTime() /
253                 CloudDbConstant::TEN_THOUSAND;
254             tableData.extend[g_deleteField] = true;
255             tableData.extend[g_cursorField] = std::to_string(currentCursor_++);
256             LOGD("[VirtualCloudDb] DeleteByGid, gid %s", std::get<std::string>(extend[g_gidField]).c_str());
257             tableData.record.clear();
258             break;
259         }
260     }
261     return OK;
262 }
263 
Query(const std::string & tableName,VBucket & extend,std::vector<VBucket> & data)264 DBStatus VirtualCloudDb::Query(const std::string &tableName, VBucket &extend, std::vector<VBucket> &data)
265 {
266     LOGW("begin query %s", tableName.c_str());
267     if (actionStatus_ != OK) {
268         return actionStatus_;
269     }
270     if (cloudError_) {
271         return DB_ERROR;
272     }
273     if (blockTimeMs_ != 0) {
274         std::this_thread::sleep_for(std::chrono::milliseconds(blockTimeMs_));
275     }
276     if (forkFunc_) {
277         forkFunc_(tableName, extend);
278     }
279     if (queryTimes_.find(tableName) == queryTimes_.end()) {
280         queryTimes_.try_emplace(tableName, 0);
281     }
282     queryTimes_[tableName]++;
283     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
284     if (cloudData_.find(tableName) == cloudData_.end()) {
285         return QUERY_END;
286     }
287     std::string cursor = std::get<std::string>(extend[g_cursorField]);
288     bool isIncreCursor = (cursor.substr(0, increPrefix_.size()) == increPrefix_);
289     LOGD("extend size: %zu type: %zu  expect: %zu, cursor: %s", extend.size(), extend[g_cursorField].index(),
290         TYPE_INDEX<std::string>, cursor.c_str());
291     if (isIncreCursor) {
292         GetCloudData(cursor, isIncreCursor, incrementCloudData_[tableName], data, extend);
293     } else {
294         cursor = cursor.empty() ? "0" : cursor;
295         GetCloudData(cursor, isIncreCursor, cloudData_[tableName], data, extend);
296         if (data.empty()) {
297             extend[g_cursorField] = std::to_string(currentCursor_ - 1);
298         }
299     }
300     if (!isIncreCursor && data.empty() && isSetCrementCloudData_) {
301         extend[g_cursorField] = increPrefix_;
302         return OK;
303     }
304     return (data.empty() || data.size() < static_cast<size_t>(queryLimit_)) ? QUERY_END : OK;
305 }
306 
GetCloudData(const std::string & cursor,bool isIncreCursor,std::vector<CloudData> allData,std::vector<VBucket> & data,VBucket & extend)307 void VirtualCloudDb::GetCloudData(const std::string &cursor, bool isIncreCursor, std::vector<CloudData> allData,
308     std::vector<VBucket> &data, VBucket &extend)
309 {
310     std::vector<QueryNode> queryNodes;
311     auto it = extend.find(g_queryField);
312     if (it != extend.end()) {
313         Bytes bytes = std::get<Bytes>(extend[g_queryField]);
314         DBStatus status = OK;
315         queryNodes = RelationalStoreManager::ParserQueryNodes(bytes, status);
316     }
317     for (auto &tableData : allData) {
318         std::string srcCursor = std::get<std::string>(tableData.extend[g_cursorField]);
319         if ((!isIncreCursor && std::stol(srcCursor) > std::stol(cursor)) || isIncreCursor) {
320             if ((!queryNodes.empty()) && (!IsCloudGidMatching(queryNodes, tableData.extend)) &&
321                 (!IsPrimaryKeyMatching(queryNodes, tableData.record))) {
322                 continue;
323             }
324             VBucket bucket = tableData.record;
325             for (const auto &ex: tableData.extend) {
326                 bucket.insert(ex);
327             }
328             data.push_back(std::move(bucket));
329             if (!isIncreCursor) {
330                 extend[g_cursorField] = srcCursor;
331             }
332         }
333         if (data.size() >= static_cast<size_t>(queryLimit_)) {
334             return;
335         }
336     }
337 }
338 
IsPrimaryKeyMatching(const std::vector<QueryNode> & queryNodes,VBucket & record)339 bool VirtualCloudDb::IsPrimaryKeyMatching(const std::vector<QueryNode> &queryNodes, VBucket &record)
340 {
341     if (record.empty()) {
342         return false;
343     }
344     for (const auto &queryNode : queryNodes) {
345         if ((queryNode.type == QueryNodeType::IN) && (queryNode.fieldName != g_gidField)) {
346             if (IsPrimaryKeyMatchingInner(queryNode, record)) {
347                 return true;
348             }
349         }
350     }
351     return false;
352 }
353 
IsPrimaryKeyMatchingInner(const QueryNode & queryNode,VBucket & record)354 bool VirtualCloudDb::IsPrimaryKeyMatchingInner(const QueryNode &queryNode, VBucket &record)
355 {
356     for (const auto &value : queryNode.fieldValue) {
357         size_t type = record[queryNode.fieldName].index();
358         switch (type) {
359             case TYPE_INDEX<std::string>: {
360                 if (std::get<std::string>(record[queryNode.fieldName]) == std::get<std::string>(value)) {
361                     return true;
362                 }
363                 break;
364             }
365             case TYPE_INDEX<int64_t>: {
366                 if (std::get<int64_t>(record[queryNode.fieldName]) == std::get<int64_t>(value)) {
367                     return true;
368                 }
369                 break;
370             }
371             case TYPE_INDEX<double>: {
372                 if (std::get<double>(record[queryNode.fieldName]) == std::get<double>(value)) {
373                     return true;
374                 }
375                 break;
376             }
377             default:
378                 break;
379         }
380     }
381     return false;
382 }
383 
IsCloudGidMatching(const std::vector<QueryNode> & queryNodes,VBucket & extend)384 bool VirtualCloudDb::IsCloudGidMatching(const std::vector<QueryNode> &queryNodes, VBucket &extend)
385 {
386     for (const auto &queryNode : queryNodes) {
387         if ((queryNode.type == QueryNodeType::IN) && (queryNode.fieldName == g_gidField)) {
388             if (IsCloudGidMatchingInner(queryNode, extend)) {
389                 return true;
390             }
391         }
392     }
393     return false;
394 }
395 
IsCloudGidMatchingInner(const QueryNode & queryNode,VBucket & extend)396 bool VirtualCloudDb::IsCloudGidMatchingInner(const QueryNode &queryNode, VBucket &extend)
397 {
398     for (const auto &value : queryNode.fieldValue) {
399         if (std::get<std::string>(extend[g_gidField]) == std::get<std::string>(value)) {
400             return true;
401         }
402     }
403     return false;
404 }
405 
InnerUpdate(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend,bool isDelete)406 DBStatus VirtualCloudDb::InnerUpdate(const std::string &tableName, std::vector<VBucket> &&record,
407     std::vector<VBucket> &extend, bool isDelete)
408 {
409     if (record.size() != extend.size()) {
410         return DB_ERROR;
411     }
412     if (forkBeforeBatchUpdateFunc_) {
413         forkBeforeBatchUpdateFunc_(tableName, record, extend, isDelete);
414     }
415     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
416     DBStatus res = InnerUpdateWithoutLock(tableName, std::move(record), extend, isDelete);
417     if (res != OK) {
418         return res;
419     }
420     if (missingExtendCount_ > 0) {
421         extend.erase(extend.end());
422     } else if (missingExtendCount_ < 0) {
423         VBucket vBucket;
424         extend.push_back(vBucket);
425     }
426     if (isDelete) {
427         for (auto &vb: extend) {
428             for (auto &[key, value]: vb) {
429                 std::ignore = std::move(value);
430                 vb.insert_or_assign(key, value);
431             }
432         }
433     }
434     if (cloudNetworkError_) {
435         return CLOUD_NETWORK_ERROR;
436     }
437     return OK;
438 }
439 
InnerUpdateWithoutLock(const std::string & tableName,std::vector<VBucket> && record,std::vector<VBucket> & extend,bool isDelete)440 DBStatus VirtualCloudDb::InnerUpdateWithoutLock(const std::string &tableName, std::vector<VBucket> &&record,
441     std::vector<VBucket> &extend, bool isDelete)
442 {
443     DBStatus res = OK;
444     for (size_t i = 0; i < record.size(); ++i) {
445         if (extend[i].find(g_gidField) == extend[i].end()) {
446             LOGE("[VirtualCloudDb] Update data should have gid");
447             return DB_ERROR;
448         }
449         if (forkUploadConflictFunc_) {
450             DBStatus ret = forkUploadConflictFunc_(tableName, extend[i], record[i], cloudData_[tableName]);
451             if (ret != OK) {
452                 res = ret;
453                 continue;
454             }
455         }
456         if (conflictInUpload_) {
457             extend[i][CloudDbConstant::ERROR_FIELD] = static_cast<int64_t>(DBStatus::CLOUD_RECORD_EXIST_CONFLICT);
458         }
459         extend[i][g_cursorField] = std::to_string(currentCursor_++);
460         AddAssetIdForExtend(record[i], extend[i]);
461         if (forkUploadFunc_) {
462             forkUploadFunc_(tableName, extend[i]);
463         }
464         if (isDelete) {
465             extend[i][g_deleteField] = true;
466         } else {
467             extend[i][g_deleteField] = false;
468         }
469         CloudData cloudData = {
470             .record = std::move(record[i]),
471             .extend = extend[i]
472         };
473         extend[i][CloudDbConstant::VERSION_FIELD] = std::to_string(currentVersion_++);
474         DBStatus ret = UpdateCloudData(tableName, std::move(cloudData));
475         if (ret == CLOUD_VERSION_CONFLICT) {
476             extend[i][CloudDbConstant::ERROR_FIELD] = static_cast<int64_t>(DBStatus::CLOUD_RECORD_EXIST_CONFLICT);
477             res = CLOUD_VERSION_CONFLICT;
478         } else if (ret != OK) {
479             return ret;
480         }
481     }
482     return res;
483 }
484 
UpdateCloudData(const std::string & tableName,VirtualCloudDb::CloudData && cloudData)485 DBStatus VirtualCloudDb::UpdateCloudData(const std::string &tableName, VirtualCloudDb::CloudData &&cloudData)
486 {
487     if (cloudData_.find(tableName) == cloudData_.end()) {
488         LOGE("[VirtualCloudDb] update cloud data failed, not found tableName %s", tableName.c_str());
489         return DB_ERROR;
490     }
491     std::string paramGid = std::get<std::string>(cloudData.extend[g_gidField]);
492     bool paramDelete = std::get<bool>(cloudData.extend[g_deleteField]);
493     for (auto &data: cloudData_[tableName]) {
494         std::string srcGid = std::get<std::string>(data.extend[g_gidField]);
495         if (srcGid != paramGid) {
496             continue;
497         }
498         if (paramDelete) {
499             if (data.extend.find(g_deleteField) != data.extend.end() &&
500                 std::get<bool>(data.extend[g_deleteField])) {
501                 LOGE("[VirtualCloudDb] current data has been delete gid %s", paramGid.c_str());
502                 return DB_ERROR;
503             }
504             LOGD("[VirtualCloudDb] delete data, gid %s", paramGid.c_str());
505         }
506         if (cloudData.extend.find(CloudDbConstant::VERSION_FIELD) != cloudData.extend.end()) {
507             if (std::get<std::string>(data.extend[CloudDbConstant::VERSION_FIELD]) !=
508                 std::get<std::string>(cloudData.extend[CloudDbConstant::VERSION_FIELD])) {
509                 return CLOUD_VERSION_CONFLICT;
510             }
511         }
512         cloudData.extend[CloudDbConstant::VERSION_FIELD] = std::to_string(currentVersion_ - 1);
513         data = std::move(cloudData);
514         return OK;
515     }
516     LOGE("[VirtualCloudDb] update cloud data failed, not found gid %s", paramGid.c_str());
517     return DB_ERROR;
518 }
519 
SetCloudError(bool cloudError)520 void VirtualCloudDb::SetCloudError(bool cloudError)
521 {
522     cloudError_ = cloudError;
523 }
524 
SetBlockTime(int32_t blockTime)525 void VirtualCloudDb::SetBlockTime(int32_t blockTime)
526 {
527     blockTimeMs_ = blockTime;
528 }
529 
ClearHeartbeatCount()530 void VirtualCloudDb::ClearHeartbeatCount()
531 {
532     heartbeatCount_ = 0;
533 }
534 
GetHeartbeatCount() const535 int32_t VirtualCloudDb::GetHeartbeatCount() const
536 {
537     return heartbeatCount_;
538 }
539 
GetLockStatus() const540 bool VirtualCloudDb::GetLockStatus() const
541 {
542     return lockStatus_;
543 }
544 
SetHeartbeatError(bool heartbeatError)545 void VirtualCloudDb::SetHeartbeatError(bool heartbeatError)
546 {
547     heartbeatError_ = heartbeatError;
548 }
549 
SetInsertHook(const std::function<void (VBucket &)> & insertCheckFunc)550 void VirtualCloudDb::SetInsertHook(const std::function<void(VBucket &)> &insertCheckFunc)
551 {
552     insertCheckFunc_ = insertCheckFunc;
553 }
554 
SetIncrementData(const std::string & tableName,const VBucket & record,const VBucket & extend)555 void VirtualCloudDb::SetIncrementData(const std::string &tableName, const VBucket &record, const VBucket &extend)
556 {
557     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
558     isSetCrementCloudData_ = true;
559     auto iter = incrementCloudData_.find(tableName);
560     if (iter == incrementCloudData_.end()) {
561         return;
562     }
563     CloudData data = {record, extend};
564     iter->second.push_back(data);
565 }
566 
GetQueryTimes(const std::string & tableName)567 uint32_t VirtualCloudDb::GetQueryTimes(const std::string &tableName)
568 {
569     if (queryTimes_.find(tableName) == queryTimes_.end()) {
570         return 0;
571     }
572     return queryTimes_[tableName];
573 }
574 
SetActionStatus(DBStatus status)575 void VirtualCloudDb::SetActionStatus(DBStatus status)
576 {
577     actionStatus_ = status;
578 }
579 
GetDataStatus(const std::string & gid,bool & deleteStatus)580 DBStatus VirtualCloudDb::GetDataStatus(const std::string &gid, bool &deleteStatus)
581 {
582     for (const auto &[tableName, tableDataList]: cloudData_) {
583         for (auto &tableData : tableDataList) {
584             if (std::get<std::string>(tableData.extend.at(g_gidField)) == gid) {
585                 deleteStatus = std::get<bool>(tableData.extend.at(g_deleteField));
586                 LOGI("tableName %s gid %s deleteStatus is %d", tableName.c_str(), gid.c_str(), deleteStatus);
587                 return OK;
588             }
589         }
590     }
591     LOGE("not found gid %s ", gid.c_str());
592     return NOT_FOUND;
593 }
594 
ClearAllData()595 void VirtualCloudDb::ClearAllData()
596 {
597     std::lock_guard<std::mutex> autoLock(cloudDataMutex_);
598     cloudData_.clear();
599     incrementCloudData_.clear();
600     queryTimes_.clear();
601 }
602 
ForkQuery(const std::function<void (const std::string &,VBucket &)> & forkFunc)603 void VirtualCloudDb::ForkQuery(const std::function<void(const std::string &, VBucket &)> &forkFunc)
604 {
605     forkFunc_ = forkFunc;
606 }
607 
ForkUpload(const std::function<void (const std::string &,VBucket &)> & forkUploadFunc)608 void VirtualCloudDb::ForkUpload(const std::function<void(const std::string &, VBucket &)> &forkUploadFunc)
609 {
610     forkUploadFunc_ = forkUploadFunc;
611 }
612 
ForkBeforeBatchUpdate(const std::function<void (const std::string &,std::vector<VBucket> &,std::vector<VBucket> &,bool isDelete)> & forkBeforeBatchUpdateFunc)613 void VirtualCloudDb::ForkBeforeBatchUpdate(const std::function<void(const std::string &, std::vector<VBucket> &,
614     std::vector<VBucket> &, bool isDelete)> &forkBeforeBatchUpdateFunc)
615 {
616     forkBeforeBatchUpdateFunc_ = forkBeforeBatchUpdateFunc;
617 }
618 
GetLockCount() const619 int32_t VirtualCloudDb::GetLockCount() const
620 {
621     return lockCount_;
622 }
623 
Reset()624 void VirtualCloudDb::Reset()
625 {
626     lockCount_ = 0;
627 }
628 
SetInsertFailed(int32_t count)629 void VirtualCloudDb::SetInsertFailed(int32_t count)
630 {
631     insertFailedCount_ = count;
632 }
633 
SetClearExtend(int32_t count)634 void VirtualCloudDb::SetClearExtend(int32_t count)
635 {
636     missingExtendCount_ = count;
637 }
638 
SetCloudNetworkError(bool cloudNetworkError)639 void VirtualCloudDb::SetCloudNetworkError(bool cloudNetworkError)
640 {
641     cloudNetworkError_ = cloudNetworkError;
642 }
643 
SetConflictInUpload(bool conflict)644 void VirtualCloudDb::SetConflictInUpload(bool conflict)
645 {
646     conflictInUpload_ = conflict;
647 }
648 
AddAssetIdForExtend(VBucket & record,VBucket & extend)649 void VirtualCloudDb::AddAssetIdForExtend(VBucket &record, VBucket &extend)
650 {
651     for (auto &recordData : record) {
652         if (recordData.second.index() == TYPE_INDEX<Asset>) {
653             auto &asset = std::get<Asset>(recordData.second);
654             if (asset.flag == static_cast<uint32_t>(DistributedDB::AssetOpType::INSERT)) {
655                 asset.assetId = "10";
656             }
657             extend[recordData.first] = asset;
658         }
659         if (recordData.second.index() == TYPE_INDEX<Assets>) {
660             auto &assets = std::get<Assets>(recordData.second);
661             AddAssetsIdInner(assets);
662             extend[recordData.first] = assets;
663         }
664     }
665 }
666 
AddAssetsIdInner(Assets & assets)667 void VirtualCloudDb::AddAssetsIdInner(Assets &assets)
668 {
669     for (auto &asset : assets) {
670         if (asset.flag == static_cast<uint32_t>(DistributedDB::AssetOpType::INSERT)) {
671             asset.assetId = "10";
672         }
673     }
674 }
675 
SetHeartbeatBlockTime(int32_t blockTime)676 void VirtualCloudDb::SetHeartbeatBlockTime(int32_t blockTime)
677 {
678     heartbeatBlockTimeMs_ = blockTime;
679 }
680 
ForkInsertConflict(const std::function<DBStatus (const std::string &,VBucket &,VBucket &,std::vector<CloudData> &)> & forkUploadFunc)681 void VirtualCloudDb::ForkInsertConflict(const std::function<DBStatus(const std::string &, VBucket &, VBucket &,
682     std::vector<CloudData> &)> &forkUploadFunc)
683 {
684     forkUploadConflictFunc_ = forkUploadFunc;
685 }
686 }