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