1 /*
2 * Copyright (c) 2021 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
16 #include "db_common.h"
17
18 #include <atomic>
19 #include <charconv>
20 #include <climits>
21 #include <cstdio>
22 #ifndef _WIN32
23 #include <dlfcn.h>
24 #endif
25 #include <mutex>
26 #include <queue>
27
28 #include "cloud/cloud_db_constant.h"
29 #include "cloud/cloud_db_types.h"
30 #include "db_errno.h"
31 #include "platform_specific.h"
32 #include "query_sync_object.h"
33 #include "hash.h"
34 #include "runtime_context.h"
35 #include "value_hash_calc.h"
36
37 namespace DistributedDB {
38 namespace {
39 constexpr const int32_t HEAD_SIZE = 3;
40 constexpr const int32_t END_SIZE = 3;
41 constexpr const int32_t MIN_SIZE = HEAD_SIZE + END_SIZE + 3;
42 constexpr const char *REPLACE_CHAIN = "***";
43 constexpr const char *DEFAULT_ANONYMOUS = "******";
44
RemoveFiles(const std::list<OS::FileAttr> & fileList,OS::FileType type)45 void RemoveFiles(const std::list<OS::FileAttr> &fileList, OS::FileType type)
46 {
47 for (const auto &item : fileList) {
48 if (item.fileType != type) {
49 continue;
50 }
51 int errCode = OS::RemoveFile(item.fileName);
52 if (errCode != E_OK) {
53 LOGE("Remove file failed:%d", errno);
54 }
55 }
56 }
57
RemoveDirectories(const std::list<OS::FileAttr> & fileList,OS::FileType type)58 void RemoveDirectories(const std::list<OS::FileAttr> &fileList, OS::FileType type)
59 {
60 for (auto item = fileList.rbegin(); item != fileList.rend(); ++item) {
61 if (item->fileType != type) {
62 continue;
63 }
64 int errCode = OS::RemoveDBDirectory(item->fileName);
65 if (errCode != 0) {
66 LOGE("Remove directory failed:%d", errno);
67 }
68 }
69 }
70 const std::string HEX_CHAR_MAP = "0123456789abcdef";
71 const std::string CAP_HEX_CHAR_MAP = "0123456789ABCDEF";
72 }
73
74 static std::atomic_bool g_isGrdLoaded = false;
75
CreateDirectory(const std::string & directory)76 int DBCommon::CreateDirectory(const std::string &directory)
77 {
78 bool isExisted = OS::CheckPathExistence(directory);
79 if (!isExisted) {
80 int errCode = OS::MakeDBDirectory(directory);
81 if (errCode != E_OK) {
82 return errCode;
83 }
84 }
85 return E_OK;
86 }
87
StringToVector(const std::string & src,std::vector<uint8_t> & dst)88 void DBCommon::StringToVector(const std::string &src, std::vector<uint8_t> &dst)
89 {
90 dst.resize(src.size());
91 dst.assign(src.begin(), src.end());
92 }
93
VectorToString(const std::vector<uint8_t> & src,std::string & dst)94 void DBCommon::VectorToString(const std::vector<uint8_t> &src, std::string &dst)
95 {
96 dst.clear();
97 dst.assign(src.begin(), src.end());
98 }
99
VectorToHexString(const std::vector<uint8_t> & inVec,const std::string & separator)100 std::string DBCommon::VectorToHexString(const std::vector<uint8_t> &inVec, const std::string &separator)
101 {
102 std::string outString;
103 for (auto &entry : inVec) {
104 outString.push_back(CAP_HEX_CHAR_MAP[entry >> 4]); // high 4 bits to one hex.
105 outString.push_back(CAP_HEX_CHAR_MAP[entry & 0x0F]); // low 4 bits to one hex.
106 outString += separator;
107 }
108 outString.erase(outString.size() - separator.size(), separator.size()); // remove needless separator at last
109 return outString;
110 }
111
PrintHexVector(const std::vector<uint8_t> & data,int line,const std::string & tag)112 void DBCommon::PrintHexVector(const std::vector<uint8_t> &data, int line, const std::string &tag)
113 {
114 const size_t maxDataLength = 1024;
115 const int byteHexNum = 2;
116 size_t dataLength = data.size();
117
118 if (data.size() > maxDataLength) {
119 dataLength = maxDataLength;
120 }
121
122 char *buff = new (std::nothrow) char[dataLength * byteHexNum + 1]; // dual and add one for the end;
123 if (buff == nullptr) {
124 return;
125 }
126
127 for (std::vector<uint8_t>::size_type i = 0; i < dataLength; ++i) {
128 buff[byteHexNum * i] = CAP_HEX_CHAR_MAP[data[i] >> 4]; // high 4 bits to one hex.
129 buff[byteHexNum * i + 1] = CAP_HEX_CHAR_MAP[data[i] & 0x0F]; // low 4 bits to one hex.
130 }
131 buff[dataLength * byteHexNum] = '\0';
132
133 if (line == 0) {
134 LOGD("[%s] size:%zu -- %s", tag.c_str(), data.size(), buff);
135 } else {
136 LOGD("[%s][%d] size:%zu -- %s", tag.c_str(), line, data.size(), buff);
137 }
138
139 delete []buff;
140 return;
141 }
142
TransferHashString(const std::string & devName)143 std::string DBCommon::TransferHashString(const std::string &devName)
144 {
145 if (devName.empty()) {
146 return "";
147 }
148 std::vector<uint8_t> devVect(devName.begin(), devName.end());
149 std::vector<uint8_t> hashVect;
150 int errCode = CalcValueHash(devVect, hashVect);
151 if (errCode != E_OK) {
152 return "";
153 }
154
155 return std::string(hashVect.begin(), hashVect.end());
156 }
157
TransferStringToHex(const std::string & origStr)158 std::string DBCommon::TransferStringToHex(const std::string &origStr)
159 {
160 if (origStr.empty()) {
161 return "";
162 }
163
164 std::string tmp;
165 for (auto item : origStr) {
166 unsigned char currentByte = static_cast<unsigned char>(item);
167 tmp.push_back(HEX_CHAR_MAP[currentByte >> 4]); // high 4 bits to one hex.
168 tmp.push_back(HEX_CHAR_MAP[currentByte & 0x0F]); // low 4 bits to one hex.
169 }
170 return tmp;
171 }
172
CalcValueHash(const std::vector<uint8_t> & value,std::vector<uint8_t> & hashValue)173 int DBCommon::CalcValueHash(const std::vector<uint8_t> &value, std::vector<uint8_t> &hashValue)
174 {
175 ValueHashCalc hashCalc;
176 int errCode = hashCalc.Initialize();
177 if (errCode != E_OK) {
178 return -E_INTERNAL_ERROR;
179 }
180
181 errCode = hashCalc.Update(value);
182 if (errCode != E_OK) {
183 return -E_INTERNAL_ERROR;
184 }
185
186 errCode = hashCalc.GetResult(hashValue);
187 if (errCode != E_OK) {
188 return -E_INTERNAL_ERROR;
189 }
190
191 return E_OK;
192 }
193
CreateStoreDirectory(const std::string & directory,const std::string & identifierName,const std::string & subDir,bool isCreate)194 int DBCommon::CreateStoreDirectory(const std::string &directory, const std::string &identifierName,
195 const std::string &subDir, bool isCreate)
196 {
197 std::string newDir = directory;
198 if (newDir.back() != '/') {
199 newDir += "/";
200 }
201
202 newDir += identifierName;
203 if (!isCreate) {
204 if (!OS::CheckPathExistence(newDir)) {
205 LOGE("Required path does not exist and won't create.");
206 return -E_INVALID_ARGS;
207 }
208 return E_OK;
209 }
210
211 if (directory.empty()) {
212 return -E_INVALID_ARGS;
213 }
214
215 int errCode = DBCommon::CreateDirectory(newDir);
216 if (errCode != E_OK) {
217 return errCode;
218 }
219
220 newDir += ("/" + subDir);
221 return DBCommon::CreateDirectory(newDir);
222 }
223
CopyFile(const std::string & srcFile,const std::string & dstFile)224 int DBCommon::CopyFile(const std::string &srcFile, const std::string &dstFile)
225 {
226 const int copyBlockSize = 4096;
227 std::vector<uint8_t> tmpBlock(copyBlockSize, 0);
228 int errCode;
229 FILE *fileIn = fopen(srcFile.c_str(), "rb");
230 if (fileIn == nullptr) {
231 LOGE("[Common:CpFile] open the source file error:%d", errno);
232 return -E_INVALID_FILE;
233 }
234 FILE *fileOut = fopen(dstFile.c_str(), "wb");
235 if (fileOut == nullptr) {
236 LOGE("[Common:CpFile] open the target file error:%d", errno);
237 errCode = -E_INVALID_FILE;
238 goto END;
239 }
240 for (;;) {
241 size_t readSize = fread(static_cast<void *>(tmpBlock.data()), 1, copyBlockSize, fileIn);
242 if (readSize < copyBlockSize) {
243 // not end and have error.
244 if (feof(fileIn) != 0 && ferror(fileIn) != 0) {
245 LOGE("Copy the file error:%d", errno);
246 errCode = -E_SYSTEM_API_FAIL;
247 break;
248 }
249 }
250
251 if (readSize != 0) {
252 size_t writeSize = fwrite(static_cast<void *>(tmpBlock.data()), 1, readSize, fileOut);
253 if (ferror(fileOut) != 0 || writeSize != readSize) {
254 LOGE("Write the data while copy:%d", errno);
255 errCode = -E_SYSTEM_API_FAIL;
256 break;
257 }
258 }
259
260 if (feof(fileIn) != 0) {
261 errCode = E_OK;
262 break;
263 }
264 }
265
266 END:
267 if (fileIn != nullptr) {
268 (void)fclose(fileIn);
269 }
270 if (fileOut != nullptr) {
271 (void)fclose(fileOut);
272 }
273 return errCode;
274 }
275
RemoveAllFilesOfDirectory(const std::string & dir,bool isNeedRemoveDir)276 int DBCommon::RemoveAllFilesOfDirectory(const std::string &dir, bool isNeedRemoveDir)
277 {
278 std::list<OS::FileAttr> fileList;
279 bool isExisted = OS::CheckPathExistence(dir);
280 if (!isExisted) {
281 return E_OK;
282 }
283 int errCode = OS::GetFileAttrFromPath(dir, fileList, true);
284 if (errCode != E_OK) {
285 return errCode;
286 }
287
288 RemoveFiles(fileList, OS::FileType::FILE);
289 RemoveDirectories(fileList, OS::FileType::PATH);
290 if (isNeedRemoveDir) {
291 // Pay attention to the order of deleting the directory
292 if (OS::CheckPathExistence(dir) && OS::RemoveDBDirectory(dir) != 0) {
293 LOGI("Remove the directory error:%d", errno);
294 errCode = -E_SYSTEM_API_FAIL;
295 }
296 }
297
298 return errCode;
299 }
300
GenerateIdentifierId(const std::string & storeId,const std::string & appId,const std::string & userId,const std::string & subUser,int32_t instanceId)301 std::string DBCommon::GenerateIdentifierId(const std::string &storeId,
302 const std::string &appId, const std::string &userId, const std::string &subUser, int32_t instanceId)
303 {
304 std::string id = userId + "-" + appId + "-" + storeId;
305 if (instanceId != 0) {
306 id += "-" + std::to_string(instanceId);
307 }
308 if (!subUser.empty()) {
309 id += "-" + subUser;
310 }
311 return id;
312 }
313
GenerateDualTupleIdentifierId(const std::string & storeId,const std::string & appId)314 std::string DBCommon::GenerateDualTupleIdentifierId(const std::string &storeId, const std::string &appId)
315 {
316 return appId + "-" + storeId;
317 }
318
SetDatabaseIds(KvDBProperties & properties,const DbIdParam & dbIdParam)319 void DBCommon::SetDatabaseIds(KvDBProperties &properties, const DbIdParam &dbIdParam)
320 {
321 properties.SetIdentifier(dbIdParam.userId, dbIdParam.appId, dbIdParam.storeId,
322 dbIdParam.subUser, dbIdParam.instanceId);
323 std::string oriStoreDir;
324 // IDENTIFIER_DIR no need cal with instanceId and subUser
325 std::string identifier = GenerateIdentifierId(dbIdParam.storeId, dbIdParam.appId, dbIdParam.userId);
326 if (properties.GetBoolProp(KvDBProperties::CREATE_DIR_BY_STORE_ID_ONLY, false)) {
327 oriStoreDir = dbIdParam.storeId;
328 } else {
329 oriStoreDir = identifier;
330 }
331 std::string hashIdentifier = TransferHashString(identifier);
332 std::string hashDir = TransferHashString(oriStoreDir);
333 std::string hexHashDir = TransferStringToHex(hashDir);
334 properties.SetStringProp(KvDBProperties::IDENTIFIER_DIR, hexHashDir);
335 }
336
StringMasking(const std::string & oriStr,size_t remain)337 std::string DBCommon::StringMasking(const std::string &oriStr, size_t remain)
338 {
339 #ifndef DB_DEBUG_ENV
340 if (oriStr.size() > remain) {
341 return oriStr.substr(0, remain);
342 }
343 #endif
344 return oriStr;
345 }
346
StringMiddleMasking(const std::string & name)347 std::string DBCommon::StringMiddleMasking(const std::string &name)
348 {
349 if (name.length() <= HEAD_SIZE) {
350 return DEFAULT_ANONYMOUS;
351 }
352
353 if (name.length() < MIN_SIZE) {
354 return (name.substr(0, HEAD_SIZE) + REPLACE_CHAIN);
355 }
356
357 return (name.substr(0, HEAD_SIZE) + REPLACE_CHAIN + name.substr(name.length() - END_SIZE, END_SIZE));
358 }
359
GetDistributedTableName(const std::string & device,const std::string & tableName)360 std::string DBCommon::GetDistributedTableName(const std::string &device, const std::string &tableName)
361 {
362 if (!RuntimeContext::GetInstance()->ExistTranslateDevIdCallback()) {
363 return GetDistributedTableNameWithHash(device, tableName);
364 }
365 return CalDistributedTableName(device, tableName);
366 }
367
GetDistributedTableName(const std::string & device,const std::string & tableName,const StoreInfo & info)368 std::string DBCommon::GetDistributedTableName(const std::string &device, const std::string &tableName,
369 const StoreInfo &info)
370 {
371 std::string newDeviceId;
372 if (RuntimeContext::GetInstance()->TranslateDeviceId(device, info, newDeviceId) != E_OK) {
373 return GetDistributedTableNameWithHash(device, tableName);
374 }
375 return CalDistributedTableName(newDeviceId, tableName);
376 }
377
GetDistributedTableNameWithHash(const std::string & device,const std::string & tableName)378 std::string DBCommon::GetDistributedTableNameWithHash(const std::string &device, const std::string &tableName)
379 {
380 std::string deviceHashHex = DBCommon::TransferStringToHex(DBCommon::TransferHashString(device));
381 return CalDistributedTableName(deviceHashHex, tableName);
382 }
383
CalDistributedTableName(const std::string & device,const std::string & tableName)384 std::string DBCommon::CalDistributedTableName(const std::string &device, const std::string &tableName)
385 {
386 return DBConstant::RELATIONAL_PREFIX + tableName + "_" + device;
387 }
388
GetDeviceFromName(const std::string & deviceTableName,std::string & deviceHash,std::string & tableName)389 void DBCommon::GetDeviceFromName(const std::string &deviceTableName, std::string &deviceHash, std::string &tableName)
390 {
391 std::size_t found = deviceTableName.rfind('_');
392 if (found != std::string::npos && found + 1 < deviceTableName.length() &&
393 found > DBConstant::RELATIONAL_PREFIX.length()) {
394 deviceHash = deviceTableName.substr(found + 1);
395 tableName = deviceTableName.substr(DBConstant::RELATIONAL_PREFIX.length(),
396 found - DBConstant::RELATIONAL_PREFIX.length());
397 }
398 }
399
TrimSpace(const std::string & input)400 std::string DBCommon::TrimSpace(const std::string &input)
401 {
402 std::string res;
403 res.reserve(input.length());
404 bool isPreSpace = true;
405 for (char c : input) {
406 if (std::isspace(c)) {
407 isPreSpace = true;
408 } else {
409 if (!res.empty() && isPreSpace) {
410 res += ' ';
411 }
412 res += c;
413 isPreSpace = false;
414 }
415 }
416 res.shrink_to_fit();
417 return res;
418 }
419
RTrim(std::string & oriString)420 void DBCommon::RTrim(std::string &oriString)
421 {
422 if (oriString.empty()) {
423 return;
424 }
425 oriString.erase(oriString.find_last_not_of(" ") + 1);
426 }
427
428 namespace {
CharIn(char c,const std::string & pattern)429 bool CharIn(char c, const std::string &pattern)
430 {
431 return std::any_of(pattern.begin(), pattern.end(), [c] (char p) {
432 return c == p;
433 });
434 }
435 }
436
HasConstraint(const std::string & sql,const std::string & keyWord,const std::string & prePattern,const std::string & nextPattern)437 bool DBCommon::HasConstraint(const std::string &sql, const std::string &keyWord, const std::string &prePattern,
438 const std::string &nextPattern)
439 {
440 size_t pos = 0;
441 while ((pos = sql.find(keyWord, pos)) != std::string::npos) {
442 if (pos >= 1 && CharIn(sql[pos - 1], prePattern) && ((pos + keyWord.length() == sql.length()) ||
443 ((pos + keyWord.length() < sql.length()) && CharIn(sql[pos + keyWord.length()], nextPattern)))) {
444 return true;
445 }
446 pos++;
447 }
448 return false;
449 }
450
IsSameCipher(CipherType srcType,CipherType inputType)451 bool DBCommon::IsSameCipher(CipherType srcType, CipherType inputType)
452 {
453 // At present, the default type is AES-256-GCM.
454 // So when src is default and input is AES-256-GCM,
455 // or when src is AES-256-GCM and input is default,
456 // we think they are the same type.
457 if (srcType == inputType ||
458 ((srcType == CipherType::DEFAULT || srcType == CipherType::AES_256_GCM) &&
459 (inputType == CipherType::DEFAULT || inputType == CipherType::AES_256_GCM))) {
460 return true;
461 }
462 return false;
463 }
464
ToLowerCase(const std::string & str)465 std::string DBCommon::ToLowerCase(const std::string &str)
466 {
467 std::string res(str.length(), ' ');
468 std::transform(str.begin(), str.end(), res.begin(), ::tolower);
469 return res;
470 }
471
ToUpperCase(const std::string & str)472 std::string DBCommon::ToUpperCase(const std::string &str)
473 {
474 std::string res(str.length(), ' ');
475 std::transform(str.begin(), str.end(), res.begin(), ::toupper);
476 return res;
477 }
478
CaseInsensitiveCompare(const std::string & first,const std::string & second)479 bool DBCommon::CaseInsensitiveCompare(const std::string &first, const std::string &second)
480 {
481 return (strcasecmp(first.c_str(), second.c_str()) == 0);
482 }
483
CheckIsAlnumOrUnderscore(const std::string & text)484 bool DBCommon::CheckIsAlnumOrUnderscore(const std::string &text)
485 {
486 auto iter = std::find_if_not(text.begin(), text.end(), [](char c) {
487 return (std::isalnum(c) || c == '_');
488 });
489 return iter == text.end();
490 }
491
CheckQueryWithoutMultiTable(const Query & query)492 bool DBCommon::CheckQueryWithoutMultiTable(const Query &query)
493 {
494 QuerySyncObject syncObject(query);
495 if (!syncObject.GetRelationTableNames().empty()) {
496 LOGE("check query table names from tables failed!");
497 return false;
498 }
499 if (!QuerySyncObject::GetQuerySyncObject(query).empty()) {
500 LOGE("check query object from table failed!");
501 return false;
502 }
503 return true;
504 }
505
506 /* this function us topology sorting algorithm to detect whether a ring exists in the dependency
507 * the algorithm main procedure as below:
508 * 1. select a point which in-degree is 0 in the graph and record it;
509 * 2. delete the point and all edges starting from this point;
510 * 3. repeat step 1 and 2, until the graph is empty or there is no point with a zero degree
511 * */
IsCircularDependency(int size,const std::vector<std::vector<int>> & dependency)512 bool DBCommon::IsCircularDependency(int size, const std::vector<std::vector<int>> &dependency)
513 {
514 std::vector<int> inDegree(size, 0); // save in-degree of every point
515 std::vector<std::vector<int>> adjacencyList(size);
516 for (size_t i = 0; i < dependency.size(); i++) {
517 adjacencyList[dependency[i][0]].push_back(dependency[i][1]); // update adjacencyList
518 inDegree[dependency[i][1]]++;
519 }
520 std::queue<int> que;
521 for (size_t i = 0; i < inDegree.size(); i++) {
522 if (inDegree[i] == 0) {
523 que.push(i); // push all point which in-degree = 0
524 }
525 }
526
527 int zeroDegreeCnt = static_cast<int>(que.size());
528 while (!que.empty()) {
529 int index = que.front();
530 que.pop();
531 for (size_t i = 0; i < adjacencyList[index].size(); ++i) {
532 int j = adjacencyList[index][i]; // adjacencyList[index] save the point which is connected to index
533 inDegree[j]--;
534 if (inDegree[j] == 0) {
535 zeroDegreeCnt++;
536 que.push(j);
537 }
538 }
539 }
540 return zeroDegreeCnt != size;
541 }
542
SerializeWaterMark(Timestamp localMark,const std::string & cloudMark,Value & blobMeta)543 int DBCommon::SerializeWaterMark(Timestamp localMark, const std::string &cloudMark, Value &blobMeta)
544 {
545 uint64_t length = Parcel::GetUInt64Len() + Parcel::GetStringLen(cloudMark);
546 blobMeta.resize(length);
547 Parcel parcel(blobMeta.data(), blobMeta.size());
548 parcel.WriteUInt64(localMark);
549 parcel.WriteString(cloudMark);
550 if (parcel.IsError()) {
551 LOGE("[DBCommon] Parcel error while serializing cloud meta data.");
552 return -E_PARSE_FAIL;
553 }
554 return E_OK;
555 }
556
GetPrefixTableName(const TableName & tableName)557 Key DBCommon::GetPrefixTableName(const TableName &tableName)
558 {
559 TableName newName = CloudDbConstant::CLOUD_META_TABLE_PREFIX + tableName;
560 Key prefixedTableName(newName.begin(), newName.end());
561 return prefixedTableName;
562 }
563
InsertNodesByScore(const std::map<std::string,std::map<std::string,bool>> & graph,const std::vector<std::string> & generateNodes,const std::map<std::string,int> & scoreGraph,std::list<std::string> & insertTarget)564 void DBCommon::InsertNodesByScore(const std::map<std::string, std::map<std::string, bool>> &graph,
565 const std::vector<std::string> &generateNodes, const std::map<std::string, int> &scoreGraph,
566 std::list<std::string> &insertTarget)
567 {
568 auto copyGraph = graph;
569 // insert all nodes into res
570 for (const auto &generateNode : generateNodes) {
571 auto iterator = insertTarget.begin();
572 for (; iterator != insertTarget.end(); iterator++) {
573 // don't compare two no reachable node
574 if (!copyGraph[*iterator][generateNode] && !copyGraph[generateNode][*iterator]) {
575 continue;
576 }
577 if (scoreGraph.find(*iterator) == scoreGraph.end() || scoreGraph.find(generateNode) == scoreGraph.end()) {
578 // should not happen
579 LOGW("[DBCommon] not find score in graph");
580 continue;
581 }
582 if (scoreGraph.at(*iterator) <= scoreGraph.at(generateNode)) {
583 break;
584 }
585 }
586 insertTarget.insert(iterator, generateNode);
587 }
588 }
589
GenerateNodesByNodeWeight(const std::vector<std::string> & nodes,const std::map<std::string,std::map<std::string,bool>> & graph,const std::map<std::string,int> & nodeWeight)590 std::list<std::string> DBCommon::GenerateNodesByNodeWeight(const std::vector<std::string> &nodes,
591 const std::map<std::string, std::map<std::string, bool>> &graph,
592 const std::map<std::string, int> &nodeWeight)
593 {
594 std::list<std::string> res;
595 std::set<std::string> paramNodes;
596 std::set<std::string> visitNodes;
597 for (const auto &node : nodes) {
598 res.push_back(node);
599 paramNodes.insert(node);
600 visitNodes.insert(node);
601 }
602 // find all node which can be reached by param nodes
603 for (const auto &source : paramNodes) {
604 if (graph.find(source) == graph.end()) {
605 continue;
606 }
607 for (const auto &[target, reach] : graph.at(source)) {
608 if (reach) {
609 visitNodes.insert(target);
610 }
611 }
612 }
613 std::vector<std::string> generateNodes;
614 for (const auto &node : visitNodes) {
615 // ignore the node which is param
616 if (paramNodes.find(node) == paramNodes.end()) {
617 generateNodes.push_back(node);
618 }
619 }
620 InsertNodesByScore(graph, generateNodes, nodeWeight, res);
621 return res;
622 }
623
HasPrimaryKey(const std::vector<Field> & fields)624 bool DBCommon::HasPrimaryKey(const std::vector<Field> &fields)
625 {
626 for (const auto &field : fields) {
627 if (field.primary) {
628 return true;
629 }
630 }
631 return false;
632 }
633
IsRecordError(const VBucket & record)634 bool DBCommon::IsRecordError(const VBucket &record)
635 {
636 // check record err should deal or skip, false is no error or error is considered, true is error not considered
637 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
638 return false;
639 }
640 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
641 return false;
642 }
643 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
644 return status != static_cast<int64_t>(DBStatus::CLOUD_RECORD_EXIST_CONFLICT) &&
645 status != static_cast<int64_t>(DBStatus::CLOUD_RECORD_ALREADY_EXISTED) &&
646 status != static_cast<int64_t>(DBStatus::CLOUD_RECORD_NOT_FOUND);
647 }
648
IsIntTypeRecordError(const VBucket & record)649 bool DBCommon::IsIntTypeRecordError(const VBucket &record)
650 {
651 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
652 return false;
653 }
654 return record.at(CloudDbConstant::ERROR_FIELD).index() == TYPE_INDEX<int64_t>;
655 }
656
IsRecordIgnored(const VBucket & record)657 bool DBCommon::IsRecordIgnored(const VBucket &record)
658 {
659 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
660 return false;
661 }
662 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
663 return false;
664 }
665 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
666 return status == static_cast<int64_t>(DBStatus::CLOUD_RECORD_EXIST_CONFLICT) ||
667 status == static_cast<int64_t>(DBStatus::CLOUD_VERSION_CONFLICT);
668 }
669
IsRecordVersionConflict(const VBucket & record)670 bool DBCommon::IsRecordVersionConflict(const VBucket &record)
671 {
672 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
673 return false;
674 }
675 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
676 return false;
677 }
678 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
679 return status == static_cast<int64_t>(DBStatus::CLOUD_VERSION_CONFLICT);
680 }
681
IsRecordAssetsMissing(const VBucket & record)682 bool DBCommon::IsRecordAssetsMissing(const VBucket &record)
683 {
684 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
685 return false;
686 }
687 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
688 return false;
689 }
690 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
691 return status == static_cast<int64_t>(DBStatus::LOCAL_ASSET_NOT_FOUND);
692 }
693
IsRecordDelete(const VBucket & record)694 bool DBCommon::IsRecordDelete(const VBucket &record)
695 {
696 if (record.find(CloudDbConstant::DELETE_FIELD) == record.end()) {
697 return false;
698 }
699 if (record.at(CloudDbConstant::DELETE_FIELD).index() != TYPE_INDEX<bool>) {
700 return false;
701 }
702 return std::get<bool>(record.at(CloudDbConstant::DELETE_FIELD));
703 }
704
IsCloudRecordNotFound(const VBucket & record)705 bool DBCommon::IsCloudRecordNotFound(const VBucket &record)
706 {
707 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
708 return false;
709 }
710 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
711 return false;
712 }
713 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
714 return status == static_cast<int64_t>(DBStatus::CLOUD_RECORD_NOT_FOUND);
715 }
716
IsCloudRecordAlreadyExisted(const VBucket & record)717 bool DBCommon::IsCloudRecordAlreadyExisted(const VBucket &record)
718 {
719 if (record.find(CloudDbConstant::ERROR_FIELD) == record.end()) {
720 return false;
721 }
722 if (record.at(CloudDbConstant::ERROR_FIELD).index() != TYPE_INDEX<int64_t>) {
723 return false;
724 }
725 auto status = std::get<int64_t>(record.at(CloudDbConstant::ERROR_FIELD));
726 return status == static_cast<int64_t>(DBStatus::CLOUD_RECORD_ALREADY_EXISTED);
727 }
728
IsNeedCompensatedForUpload(const VBucket & uploadExtend,const CloudWaterType & type)729 bool DBCommon::IsNeedCompensatedForUpload(const VBucket &uploadExtend, const CloudWaterType &type)
730 {
731 return (DBCommon::IsCloudRecordAlreadyExisted(uploadExtend) && type == CloudWaterType::INSERT) ||
732 (DBCommon::IsCloudRecordNotFound(uploadExtend) && type == CloudWaterType::UPDATE);
733 }
734
IsRecordSuccess(const VBucket & record)735 bool DBCommon::IsRecordSuccess(const VBucket &record)
736 {
737 return record.find(CloudDbConstant::ERROR_FIELD) == record.end();
738 }
739
GenerateHashLabel(const DBInfo & dbInfo)740 std::string DBCommon::GenerateHashLabel(const DBInfo &dbInfo)
741 {
742 if (dbInfo.syncDualTupleMode) {
743 return DBCommon::TransferHashString(dbInfo.appId + "-" + dbInfo.storeId);
744 }
745 return DBCommon::TransferHashString(dbInfo.userId + "-" + dbInfo.appId + "-" + dbInfo.storeId);
746 }
747
EraseBit(uint64_t origin,uint64_t eraseBit)748 uint64_t DBCommon::EraseBit(uint64_t origin, uint64_t eraseBit)
749 {
750 return origin & (~eraseBit);
751 }
752
LoadGrdLib(void)753 void DBCommon::LoadGrdLib(void)
754 {
755 static std::once_flag loadOnceFlag;
756 std::call_once(loadOnceFlag, []() {
757 if (!g_isGrdLoaded) {
758 #ifndef _WIN32
759 if (dlopen("libarkdata_db_core.z.so", RTLD_LAZY) != NULL) {
760 g_isGrdLoaded = true;
761 } else {
762 LOGW("[DBCommon] unable to load grd lib, errno: %d, %s", errno, dlerror());
763 }
764 #endif
765 }
766 });
767 }
768
IsGrdLibLoaded(void)769 bool DBCommon::IsGrdLibLoaded(void)
770 {
771 return g_isGrdLoaded;
772 }
773
CheckCloudSyncConfigValid(const CloudSyncConfig & config)774 bool DBCommon::CheckCloudSyncConfigValid(const CloudSyncConfig &config)
775 {
776 if (config.maxUploadCount < CloudDbConstant::MIN_UPLOAD_BATCH_COUNT ||
777 config.maxUploadCount > CloudDbConstant::MAX_UPLOAD_BATCH_COUNT) {
778 LOGE("[DBCommon] invalid upload count %" PRId32, config.maxUploadCount);
779 return false;
780 }
781 if (config.maxUploadSize < CloudDbConstant::MIN_UPLOAD_SIZE ||
782 config.maxUploadSize > CloudDbConstant::MAX_UPLOAD_SIZE) {
783 LOGE("[DBCommon] invalid upload size %" PRId32, config.maxUploadSize);
784 return false;
785 }
786 if (config.maxRetryConflictTimes < CloudDbConstant::MIN_RETRY_CONFLICT_COUNTS) {
787 LOGE("[DBCommon] invalid retry conflict count %" PRId32, config.maxRetryConflictTimes);
788 return false;
789 }
790 return true;
791 }
792
GetCursorKey(const std::string & tableName)793 std::string DBCommon::GetCursorKey(const std::string &tableName)
794 {
795 return DBConstant::RELATIONAL_PREFIX + "cursor_" + ToLowerCase(tableName);
796 }
797
ConvertToUInt64(const std::string & str,uint64_t & value)798 bool DBCommon::ConvertToUInt64(const std::string &str, uint64_t &value)
799 {
800 auto [ptr, errCode] = std::from_chars(str.data(), str.data() + str.size(), value);
801 return errCode == std::errc{} && ptr == str.data() + str.size();
802 }
803
CmpModifyTime(const std::string & preModifyTimeStr,const std::string & curModifyTimeStr)804 bool CmpModifyTime(const std::string &preModifyTimeStr, const std::string &curModifyTimeStr)
805 {
806 uint64_t curModifyTime = 0;
807 uint64_t preModifyTime = 0;
808 if (preModifyTimeStr.empty() || !DBCommon::ConvertToUInt64(preModifyTimeStr, preModifyTime)) {
809 return true;
810 }
811 if (curModifyTimeStr.empty() || !DBCommon::ConvertToUInt64(curModifyTimeStr, curModifyTime)) {
812 return false;
813 }
814 return curModifyTime >= preModifyTime;
815 }
816
RemoveDuplicateAssetsData(std::vector<Asset> & assets)817 void DBCommon::RemoveDuplicateAssetsData(std::vector<Asset> &assets)
818 {
819 std::unordered_map<std::string, size_t> indexMap;
820 size_t vectorSize = assets.size();
821 std::vector<size_t> arr(vectorSize, 0);
822 for (std::vector<DistributedDB::Asset>::size_type i = 0; i < assets.size(); ++i) {
823 DistributedDB::Asset asset = assets.at(i);
824 auto it = indexMap.find(asset.name);
825 if (it == indexMap.end()) {
826 indexMap[asset.name] = i;
827 continue;
828 }
829 size_t prevIndex = it->second;
830 Asset &prevAsset = assets.at(prevIndex);
831 if (prevAsset.assetId.empty() && !asset.assetId.empty()) {
832 arr[prevIndex] = 1;
833 indexMap[asset.name] = i;
834 continue;
835 }
836 if (!prevAsset.assetId.empty() && asset.assetId.empty()) {
837 arr[i] = 1;
838 indexMap[asset.name] = prevIndex;
839 continue;
840 }
841 if (CmpModifyTime(prevAsset.modifyTime, asset.modifyTime)) {
842 arr[prevIndex] = 1;
843 indexMap[asset.name] = i;
844 continue;
845 }
846 arr[i] = 1;
847 indexMap[asset.name] = prevIndex;
848 }
849 indexMap.clear();
850 size_t arrIndex = 0;
851 for (auto it = assets.begin(); it != assets.end();) {
852 if (arr[arrIndex] == 1) {
853 it = assets.erase(it);
854 } else {
855 it++;
856 }
857 arrIndex++;
858 }
859 }
860 } // namespace DistributedDB
861