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 #ifndef SQLITE_SINGLE_VER_RELATIONAL_STORAGE_EXECUTOR_H 16 #define SQLITE_SINGLE_VER_RELATIONAL_STORAGE_EXECUTOR_H 17 #ifdef RELATIONAL_STORE 18 19 #include <atomic> 20 #include "cloud/asset_operation_utils.h" 21 #include "cloud/cloud_db_constant.h" 22 #include "cloud/cloud_store_types.h" 23 #include "cloud/cloud_upload_recorder.h" 24 #include "data_transformer.h" 25 #include "db_types.h" 26 #include "icloud_sync_storage_interface.h" 27 #include "macro_utils.h" 28 #include "query_object.h" 29 #include "relational_row_data.h" 30 #include "relational_store_delegate.h" 31 #include "relational_sync_data_inserter.h" 32 #include "sqlite_log_table_manager.h" 33 #include "sqlite_single_ver_relational_continue_token.h" 34 #include "sqlite_storage_executor.h" 35 #include "sqlite_utils.h" 36 #include "tracker_table.h" 37 38 namespace DistributedDB { 39 class SQLiteSingleVerRelationalStorageExecutor : public SQLiteStorageExecutor { 40 public: 41 enum class PutDataMode { 42 SYNC, 43 USER 44 }; 45 enum class MarkFlagOption { 46 DEFAULT, 47 SET_WAIT_COMPENSATED_SYNC 48 }; 49 50 struct UpdateCursorContext { 51 int errCode = E_OK; 52 uint64_t cursor; 53 }; 54 55 SQLiteSingleVerRelationalStorageExecutor(sqlite3 *dbHandle, bool writable, DistributedTableMode mode); 56 ~SQLiteSingleVerRelationalStorageExecutor() override = default; 57 58 // Delete the copy and assign constructors 59 DISABLE_COPY_ASSIGN_MOVE(SQLiteSingleVerRelationalStorageExecutor); 60 61 int ResetLogStatus(std::string &tableName); 62 63 int CreateRelationalLogTable(DistributedTableMode mode, bool isUpgraded, const std::string &identity, 64 TableInfo &table); 65 66 // The parameter "identity" is a hash string that identifies a device 67 int CreateDistributedTable(DistributedTableMode mode, bool isUpgraded, const std::string &identity, 68 TableInfo &table); 69 70 int UpgradeDistributedTable(const std::string &tableName, DistributedTableMode mode, bool &schemaChanged, 71 RelationalSchemaObject &schema, TableSyncType syncType); 72 73 int StartTransaction(TransactType type); 74 int Commit(); 75 int Rollback(); 76 77 // For Get sync data 78 int GetSyncDataByQuery(std::vector<DataItem> &dataItems, size_t appendLength, const DataSizeSpecInfo &sizeInfo, 79 std::function<int(sqlite3 *, sqlite3_stmt *&, sqlite3_stmt *&, bool &)> getStmt, const TableInfo &tableInfo); 80 81 // operation of meta data 82 int GetKvData(const Key &key, Value &value) const; 83 int PutKvData(const Key &key, const Value &value) const; 84 int DeleteMetaData(const std::vector<Key> &keys) const; 85 int DeleteMetaDataByPrefixKey(const Key &keyPrefix) const; 86 int GetAllMetaKeys(std::vector<Key> &keys) const; 87 88 // For Put sync data 89 int SaveSyncItems(RelationalSyncDataInserter &inserter, bool useTrans = true); 90 91 int CheckDBModeForRelational(); 92 93 int DeleteDistributedDeviceTable(const std::string &device, const std::string &tableName); 94 95 int DeleteDistributedAllDeviceTableLog(const std::string &tableName); 96 97 int DeleteDistributedDeviceTableLog(const std::string &device, const std::string &tableName); 98 99 int DeleteDistributedLogTable(const std::string &tableName); 100 101 int CheckAndCleanDistributedTable(const std::vector<std::string> &tableNames, 102 std::vector<std::string> &missingTables); 103 104 int CreateDistributedDeviceTable(const std::string &device, const TableInfo &baseTbl, const StoreInfo &info); 105 106 int CheckQueryObjectLegal(const TableInfo &table, QueryObject &query, const std::string &schemaVersion); 107 108 int CheckQueryObjectLegal(const QuerySyncObject &query); 109 110 int GetMaxTimestamp(const std::vector<std::string> &tablesName, Timestamp &maxTimestamp) const; 111 112 int ExecuteQueryBySqlStmt(const std::string &sql, const std::vector<std::string> &bindArgs, int packetSize, 113 std::vector<std::string> &colNames, std::vector<RelationalRowData *> &data); 114 115 int SaveSyncDataItems(RelationalSyncDataInserter &inserter); 116 117 int CheckEncryptedOrCorrupted() const; 118 119 int GetExistsDeviceList(std::set<std::string> &devices) const; 120 121 int GetUploadCount(const Timestamp ×tamp, bool isCloudForcePush, bool isCompensatedTask, 122 QuerySyncObject &query, int64_t &count); 123 124 int GetAllUploadCount(const std::vector<Timestamp> ×tampVec, bool isCloudForcePush, bool isCompensatedTask, 125 QuerySyncObject &query, int64_t &count); 126 127 int UpdateCloudLogGid(const CloudSyncData &cloudDataResult, bool ignoreEmptyGid); 128 129 int GetSyncCloudData(const CloudUploadRecorder &uploadRecorder, CloudSyncData &cloudDataResult, 130 SQLiteSingleVerRelationalContinueToken &token); 131 132 int GetSyncCloudGid(QuerySyncObject &query, const SyncTimeRange &syncTimeRange, bool isCloudForcePushStrategy, 133 bool isCompensatedTask, std::vector<std::string> &cloudGid); 134 135 void SetLocalSchema(const RelationalSchemaObject &localSchema); 136 137 int GetInfoByPrimaryKeyOrGid(const TableSchema &tableSchema, const VBucket &vBucket, 138 DataInfoWithLog &dataInfoWithLog, VBucket &assetInfo); 139 140 int PutCloudSyncData(const std::string &tableName, const TableSchema &tableSchema, 141 const TrackerTable &trackerTable, DownloadData &downloadData); 142 143 int UpdateAssetStatusForAssetOnly(const TableSchema &tableSchema, VBucket &asset); 144 145 int FillCloudAssetForDownload(const TableSchema &tableSchema, VBucket &vBucket, bool isDownloadSuccess, 146 uint64_t &currCursor); 147 148 int DoCleanInner(ClearMode mode, const std::vector<std::string> &tableNameList, 149 const RelationalSchemaObject &localSchema, std::vector<Asset> &assets, 150 std::vector<std::string> ¬ifyTableList); 151 152 int FillCloudAssetForUpload(OpType opType, const TableSchema &tableSchema, const CloudSyncBatch &data); 153 int FillCloudVersionForUpload(const OpType opType, const CloudSyncData &data); 154 155 int SetLogTriggerStatus(bool status); 156 int SetCursorIncFlag(bool flag); 157 158 int GetCursor(const std::string &tableName, uint64_t &cursor); 159 160 int AnalysisTrackerTable(const TrackerTable &trackerTable, TableInfo &tableInfo); 161 int CreateTrackerTable(const TrackerTable &trackerTable, const TableInfo &table, bool checkData); 162 int GetOrInitTrackerSchemaFromMeta(RelationalSchemaObject &schema); 163 int ExecuteSql(const SqlCondition &condition, std::vector<VBucket> &records); 164 165 int GetClearWaterMarkTables(const std::vector<TableReferenceProperty> &tableReferenceProperty, 166 const RelationalSchemaObject &schema, std::set<std::string> &clearWaterMarkTables); 167 int CreateTempSyncTrigger(const TrackerTable &trackerTable, bool flag); 168 int GetAndResetServerObserverData(const std::string &tableName, ChangeProperties &changeProperties); 169 int ClearAllTempSyncTrigger(); 170 int CleanTrackerData(const std::string &tableName, int64_t cursor, bool isOnlyTrackTable); 171 int CreateSharedTable(const TableSchema &schema); 172 int DeleteTable(const std::vector<std::string> &tableNames); 173 int UpdateSharedTable(const std::map<std::string, std::vector<Field>> &updateTableNames); 174 int AlterTableName(const std::map<std::string, std::string> &tableNames); 175 int DeleteTableTrigger(const std::string &missTable) const; 176 177 int GetReferenceGid(const std::string &tableName, const CloudSyncBatch &syncBatch, 178 const std::map<std::string, std::vector<TableReferenceProperty>> &tableReference, 179 std::map<int64_t, Entries> &referenceGid); 180 181 int CheckIfExistUserTable(const std::string &tableName); 182 183 void SetLogicDelete(bool isLogicDelete); 184 int RenewTableTrigger(DistributedTableMode mode, const TableInfo &tableInfo, TableSyncType syncType); 185 186 std::pair<int, uint32_t> GetAssetsByGidOrHashKey(const TableSchema &tableSchema, const std::string &gid, 187 const Bytes &hashKey, VBucket &assets); 188 189 int FillHandleWithOpType(const OpType opType, const CloudSyncData &data, bool fillAsset, bool ignoreEmptyGid, 190 const TableSchema &tableSchema); 191 192 void SetIAssetLoader(const std::shared_ptr<IAssetLoader> &loader); 193 194 int CleanResourceForDroppedTable(const std::string &tableName); 195 196 void SetPutDataMode(PutDataMode mode); 197 198 void SetMarkFlagOption(MarkFlagOption option); 199 200 int UpgradedLogForExistedData(TableInfo &tableInfo, bool schemaChanged); 201 202 int UpdateRecordFlag(const std::string &tableName, const std::string &sql, const LogInfo &logInfo); 203 204 void MarkFlagAsUploadFinished(const std::string &tableName, const Key &hashKey, Timestamp timestamp, 205 bool isExistAssetsDownload); 206 207 int CleanUploadFinishedFlag(const std::string &tableName); 208 209 int GetWaitCompensatedSyncDataPk(const TableSchema &table, std::vector<VBucket> &data, 210 bool isQueryDownloadRecords); 211 212 int ClearUnLockingStatus(const std::string &tableName); 213 214 int MarkFlagAsConsistent(const std::string &tableName, const DownloadData &downloadData, 215 const std::set<std::string> &gidFilters); 216 217 int MarkFlagAsAssetAsyncDownload(const std::string &tableName, const DownloadData &downloadData, 218 const std::set<std::string> &gidFilters); 219 220 int CheckInventoryData(const std::string &tableName); 221 222 int UpdateRecordStatus(const std::string &tableName, const std::string &status, const Key &hashKey); 223 224 void SetUploadConfig(int32_t maxUploadCount, int32_t maxUploadSize); 225 226 int InitCursorToMeta(const std::string &tableName); 227 228 void SetTableSchema(const TableSchema &tableSchema); 229 230 int ReviseLocalModTime(const std::string &tableName, const std::vector<ReviseModTimeInfo> &revisedData); 231 232 int GetLocalDataCount(const std::string &tableName, int &dataCount, int &logicDeleteDataCount); 233 234 int GetCloudDataCount(const std::string &tableName, DownloadData &downloadData, int64_t &count); 235 236 int UpdateExtendField(const std::string &tableName, const std::set<std::string> &extendColNames); 237 238 int UpdateDeleteDataExtendField(const std::string &tableName, const std::string &lowVersionExtendColName, 239 const std::set<std::string> &oldExtendColNames, const std::set<std::string> &extendColNames); 240 241 int GetDownloadingAssetsCount(const TableSchema &tableSchema, int32_t &totalCount); 242 243 int GetDownloadingCount(const std::string &tableName, int32_t &count); 244 245 int GetDownloadAssetGid(const TableSchema &tableSchema, std::vector<std::string> &gids, 246 int64_t beginTime = 0, bool abortWithLimit = false); 247 248 int GetDownloadAssetRecordsInner(const TableSchema &tableSchema, int64_t beginTime, std::vector<std::string> &gids); 249 250 int GetDownloadAssetRecordsByGid(const TableSchema &tableSchema, const std::string gid, 251 std::vector<VBucket> &assets); 252 253 int CleanDownloadingFlag(const std::string &tableName); 254 255 int CleanDownloadingFlagByGid(const std::string &tableName, const std::string &gid, VBucket dbAssets); 256 257 void CheckAndCreateTrigger(const TableInfo &table); 258 259 int GetLockStatusByGid(const std::string &tableName, const std::string &gid, LockStatus &status); 260 261 int CompareSchemaTableColumns(const std::string &tableName); 262 263 int UpdateHashKey(DistributedTableMode mode, const TableInfo &tableInfo, TableSyncType syncType); 264 265 void SetTableMode(DistributedTableMode mode); 266 private: 267 int DoCleanLogs(const std::vector<std::string> &tableNameList, const RelationalSchemaObject &localSchema); 268 269 int DoCleanLogAndData(const std::vector<std::string> &tableNameList, 270 const RelationalSchemaObject &localSchema, std::vector<Asset> &assets); 271 272 int CleanCloudDataOnLogTable(const std::string &logTableName, ClearMode mode); 273 274 int CleanCloudDataAndLogOnUserTable(const std::string &tableName, const std::string &logTableName, 275 const RelationalSchemaObject &localSchema); 276 277 int ChangeCloudDataFlagOnLogTable(const std::string &logTableName); 278 279 int SetDataOnUserTableWithLogicDelete(const std::string &tableName, const std::string &logTableName); 280 281 static void UpdateCursor(sqlite3_context *ctx, int argc, sqlite3_value **argv); 282 283 int CreateFuncUpdateCursor(UpdateCursorContext &context, 284 void(*updateCursorFunc)(sqlite3_context *ctx, int argc, sqlite3_value **argv)) const; 285 286 int SetCursor(const std::string &tableName, uint64_t cursor); 287 288 int IncreaseCursorOnAssetData(const std::string &tableName, const std::string &gid, uint64_t &currCursor); 289 290 int GetCleanCloudDataKeys(const std::string &logTableName, std::vector<int64_t> &dataKeys, 291 bool distinguishCloudFlag); 292 293 int GetCloudAssets(const std::string &tableName, const std::vector<FieldInfo> &fieldInfos, 294 const std::vector<int64_t> &dataKeys, std::vector<Asset> &assets); 295 296 int GetAssetOnTable(const std::string &tableName, const std::string &fieldName, 297 const std::vector<int64_t> &dataKeys, std::vector<Asset> &assets); 298 299 int GetCloudAssetsOnTable(const std::string &tableName, const std::string &fieldName, 300 const std::vector<int64_t> &dataKeys, std::vector<Asset> &assets); 301 302 int GetDataItemForSync(sqlite3_stmt *statement, DataItem &dataItem, bool isGettingDeletedData) const; 303 304 int CheckDataConflictDefeated(const DataItem &item, sqlite3_stmt *queryStmt, 305 bool &isDefeated, bool &isExist, int64_t &rowId); 306 307 int SaveSyncDataItem(RelationalSyncDataInserter &inserter, SaveSyncDataStmt &saveStmt, DataItem &item); 308 309 int SaveSyncDataItem(const DataItem &dataItem, bool isUpdate, SaveSyncDataStmt &saveStmt, 310 RelationalSyncDataInserter &inserter, int64_t &rowid); 311 312 int DeleteSyncDataItem(const DataItem &dataItem, RelationalSyncDataInserter &inserter, sqlite3_stmt *&rmDataStmt); 313 314 int AlterAuxTableForUpgrade(const TableInfo &oldTableInfo, const TableInfo &newTableInfo); 315 316 int DeleteSyncLog(const DataItem &item, RelationalSyncDataInserter &inserter, sqlite3_stmt *&rmLogStmt); 317 int ProcessMissQueryData(const DataItem &item, RelationalSyncDataInserter &inserter, sqlite3_stmt *&rmDataStmt, 318 sqlite3_stmt *&rmLogStmt); 319 int GetMissQueryData(sqlite3_stmt *fullStmt, DataItem &item); 320 int GetQueryDataAndStepNext(bool isFirstTime, bool isGettingDeletedData, sqlite3_stmt *queryStmt, DataItem &item, 321 Timestamp &queryTime); 322 int GetMissQueryDataAndStepNext(sqlite3_stmt *fullStmt, DataItem &item, Timestamp &missQueryTime); 323 324 void SetTableInfo(const TableInfo &tableInfo); // When put or get sync data, must call the func first. 325 326 int GeneLogInfoForExistedData(sqlite3 *db, const std::string &identity, const TableInfo &tableInfo, 327 std::unique_ptr<SqliteLogTableManager> &logMgrPtr, bool isTrackerTable); 328 329 int GeneLogInfoForExistedDataInner(sqlite3 *db, const std::string &identity, const TableInfo &tableInfo, 330 std::unique_ptr<SqliteLogTableManager> &logMgrPtr, bool isTrackerTable); 331 332 int CleanExtendAndCursorForDeleteData(const std::string &tableName); 333 334 int GetCloudDataForSync(const CloudUploadRecorder &uploadRecorder, sqlite3_stmt *statement, 335 CloudSyncData &cloudDataResult, uint32_t &stepNum, uint32_t &totalSize); 336 337 int PutVBucketByType(VBucket &vBucket, const Field &field, Type &cloudValue); 338 339 int GetDownloadAsset(std::vector<VBucket> &assetsV, const Field &field, Type &cloudValue); 340 341 int ExecutePutCloudData(const std::string &tableName, const TableSchema &tableSchema, 342 const TrackerTable &trackerTable, DownloadData &downloadData, std::map<int, int> &statisticMap); 343 344 std::string GetInsertSqlForCloudSync(const TableSchema &tableSchema); 345 346 int GetPrimaryKeyHashValue(const VBucket &vBucket, const TableSchema &tableSchema, std::vector<uint8_t> &hashValue, 347 bool allowEmpty = false); 348 349 int GetQueryLogStatement(const TableSchema &tableSchema, const VBucket &vBucket, const std::string &querySql, 350 const Key &hashKey, sqlite3_stmt *&selectStmt); 351 352 int GetQueryLogSql(const std::string &tableName, const VBucket &vBucket, const std::set<std::string> &pkSet, 353 std::string &querySql); 354 355 int GetQueryInfoSql(const std::string &tableName, const VBucket &vBucket, std::set<std::string> &pkSet, 356 std::vector<Field> &assetFields, std::string &querySql); 357 358 int GetFillDownloadAssetStatement(const std::string &tableName, const VBucket &vBucket, 359 const std::vector<Field> &fields, sqlite3_stmt *&statement); 360 361 int InitFillUploadAssetStatement(OpType opType, const TableSchema &tableSchema, const CloudSyncBatch &data, 362 const int &index, sqlite3_stmt *&statement); 363 364 int GetLogInfoByStatement(sqlite3_stmt *statement, LogInfo &logInfo); 365 366 int GetInfoByStatement(sqlite3_stmt *statement, const std::vector<Field> &assetFields, 367 const std::map<std::string, Field> &pkMap, DataInfoWithLog &dataInfoWithLog, VBucket &assetInfo); 368 369 int InsertCloudData(VBucket &vBucket, const TableSchema &tableSchema, const TrackerTable &trackerTable, 370 int64_t dataKey); 371 372 int InsertLogRecord(const TableSchema &tableSchema, const TrackerTable &trackerTable, VBucket &vBucket); 373 374 int BindOneField(int index, const VBucket &vBucket, const Field &field, sqlite3_stmt *updateStmt); 375 376 int BindValueToUpsertStatement(const VBucket &vBucket, const std::vector<Field> &fields, sqlite3_stmt *upsertStmt); 377 378 int BindStatusSubQueryHashKeyStatement(sqlite3_stmt *insertLogStmt, std::vector<uint8_t> &hashKey, int &index); 379 380 int BindHashKeyAndGidToInsertLogStatement(const VBucket &vBucket, const TableSchema &tableSchema, 381 const TrackerTable &trackerTable, sqlite3_stmt *insertLogStmt, int &index); 382 383 int BindShareValueToInsertLogStatement(const VBucket &vBucket, const TableSchema &tableSchema, 384 sqlite3_stmt *insertLogStmt, int &index); 385 386 int BindValueToInsertLogStatement(VBucket &vBucket, const TableSchema &tableSchema, 387 const TrackerTable &trackerTable, sqlite3_stmt *insertLogStmt); 388 389 std::string GetWhereConditionForDataTable(const std::string &gidStr, const std::set<std::string> &pkSet, 390 const std::string &tableName, bool queryByPk = true); 391 392 int GetUpdateSqlForCloudSync(const std::vector<Field> &updateFields, const TableSchema &tableSchema, 393 const std::string &gidStr, const std::set<std::string> &pkSet, std::string &updateSql); 394 395 int GetUpdateDataTableStatement(const VBucket &vBucket, const TableSchema &tableSchema, sqlite3_stmt *&updateStmt); 396 397 int UpdateCloudData(VBucket &vBucket, const TableSchema &tableSchema); 398 399 int GetUpdateLogRecordStatement(const TableSchema &tableSchema, const VBucket &vBucket, OpType opType, 400 std::vector<std::string> &updateColName, sqlite3_stmt *&updateLogStmt); 401 int AppendUpdateLogRecordWhereSqlCondition(const TableSchema &tableSchema, const VBucket &vBucket, 402 std::string &sql); 403 404 int UpdateLogRecord(const VBucket &vBucket, const TableSchema &tableSchema, OpType opType); 405 406 int BindValueToUpdateLogStatement(const VBucket &vBucket, const TableSchema &tableSchema, 407 const std::vector<std::string> &colNames, bool allowPrimaryKeyEmpty, sqlite3_stmt *updateLogStmt); 408 409 int GetDeleteStatementForCloudSync(const TableSchema &tableSchema, const std::set<std::string> &pkSet, 410 const VBucket &vBucket, sqlite3_stmt *&deleteStmt); 411 412 int UpdateTrackerTable(sqlite3 *db, const std::string &identity, const TableInfo &tableInfo, 413 std::unique_ptr<SqliteLogTableManager> &logMgrPtr, bool isTimestampOnly); 414 415 int DeleteCloudData(const std::string &tableName, const VBucket &vBucket, const TableSchema &tableSchema, 416 const TrackerTable &trackerTable); 417 418 int IsTableOnceDropped(const std::string &tableName, bool &onceDropped); 419 420 int OnlyUpdateLogTable(const VBucket &vBucket, const TableSchema &tableSchema, OpType opType); 421 422 int BindUpdateVersionStatement(const VBucket &vBucket, const Bytes &hashKey, sqlite3_stmt *&stmt); 423 int DoCleanShareTableDataAndLog(const std::vector<std::string> &tableNameList); 424 int SetDataOnShareTableWithLogicDelete(const std::string &tableName, const std::string &logTableName); 425 int CleanShareTable(const std::string &tableName); 426 427 int GetReferenceGidInner(const std::string &sourceTable, const std::string &targetTable, 428 const CloudSyncBatch &syncBatch, const std::vector<TableReferenceProperty> &targetTableReference, 429 std::map<int64_t, Entries> &referenceGid); 430 431 int GetReferenceGidByStmt(sqlite3_stmt *statement, const CloudSyncBatch &syncBatch, 432 const std::string &targetTable, std::map<int64_t, Entries> &referenceGid); 433 434 static std::string GetReferenceGidSql(const std::string &sourceTable, const std::string &targetTable, 435 const std::vector<std::string> &sourceFields, const std::vector<std::string> &targetFields); 436 437 static std::pair<std::vector<std::string>, std::vector<std::string>> SplitReferenceByField( 438 const std::vector<TableReferenceProperty> &targetTableReference); 439 440 int BindStmtWithCloudGid(const CloudSyncData &cloudDataResult, bool ignoreEmptyGid, sqlite3_stmt *&stmt); 441 442 int GetCloudDeleteSql(const std::string &table, std::string &sql); 443 444 int RemoveDataAndLog(const std::string &tableName, int64_t dataKey); 445 446 int64_t GetLocalDataKey(size_t index, const DownloadData &downloadData); 447 448 int BindStmtWithCloudGidInner(const std::string &gid, int64_t rowid, 449 sqlite3_stmt *&stmt, int &fillGidCount); 450 451 int DoCleanAssetId(const std::string &tableName, const RelationalSchemaObject &localSchema); 452 453 int CleanAssetId(const std::string &tableName, const std::vector<FieldInfo> &fieldInfos, 454 const std::vector<int64_t> &dataKeys); 455 456 int UpdateAssetIdOnUserTable(const std::string &tableName, const std::string &fieldName, 457 const std::vector<int64_t> &dataKeys, std::vector<Asset> &assets); 458 459 int GetAssetsAndUpdateAssetsId(const std::string &tableName, const std::string &fieldName, 460 const std::vector<int64_t> &dataKeys); 461 462 int CleanAssetsIdOnUserTable(const std::string &tableName, const std::string &fieldName, const int64_t rowId, 463 const std::vector<uint8_t> &assetsValue); 464 465 int InitGetAssetStmt(const std::string &sql, const std::string &gid, const Bytes &hashKey, 466 sqlite3_stmt *&stmt); 467 468 int GetAssetsByRowId(sqlite3_stmt *&selectStmt, Assets &assets); 469 470 int ExecuteFillDownloadAssetStatement(sqlite3_stmt *&stmt, int beginIndex, const std::string &cloudGid); 471 472 int CleanDownloadChangedAssets(const VBucket &vBucket, const AssetOperationUtils::RecordAssetOpType &assetOpType); 473 474 int GetAndBindFillUploadAssetStatement(const std::string &tableName, const VBucket &assets, 475 sqlite3_stmt *&statement); 476 477 int OnlyUpdateAssetId(const std::string &tableName, const TableSchema &tableSchema, const VBucket &vBucket, 478 int64_t dataKey, OpType opType); 479 480 void UpdateLocalAssetId(const VBucket &vBucket, const std::string &fieldName, Asset &asset); 481 482 void UpdateLocalAssetsId(const VBucket &vBucket, const std::string &fieldName, Assets &assets); 483 484 int BindAssetToBlobStatement(const Asset &asset, int index, sqlite3_stmt *&stmt); 485 486 int BindAssetsToBlobStatement(const Assets &assets, int index, sqlite3_stmt *&stmt); 487 488 int GetAssetInfoOnTable(sqlite3_stmt *&stmt, const std::vector<Field> &assetFields, VBucket &assetInfo); 489 490 int BindAssetFiledToBlobStatement(const TableSchema &tableSchema, const std::vector<Asset> &assetOfOneRecord, 491 const std::vector<Assets> &assetsOfOneRecord, sqlite3_stmt *&stmt); 492 493 int UpdateAssetsIdForOneRecord(const TableSchema &tableSchema, const std::string &sql, 494 const std::vector<Asset> &assetOfOneRecord, const std::vector<Assets> &assetsOfOneRecord); 495 496 bool IsNeedUpdateAssetIdInner(sqlite3_stmt *selectStmt, const VBucket &vBucket, const Field &field, 497 VBucket &assetInfo, bool &isNotIncCursor); 498 499 bool IsNeedUpdateAssetId(const TableSchema &tableSchema, int64_t dataKey, const VBucket &vBucket, 500 bool &isNotIncCursor); 501 502 int UpdateAssetId(const TableSchema &tableSchema, int64_t dataKey, const VBucket &vBucket); 503 504 int64_t GetDataFlag(); 505 506 std::string GetUpdateDataFlagSql(const VBucket &data); 507 508 std::string GetDev(); 509 510 std::vector<Field> GetUpdateField(const VBucket &vBucket, const TableSchema &tableSchema); 511 512 int GetRecordFromStmt(sqlite3_stmt *stmt, const std::vector<Field> &fields, int startIndex, VBucket &record); 513 514 int FillCloudVersionForUpload(const std::string &tableName, const CloudSyncBatch &batchData); 515 516 int QueryCount(const std::string &tableName, int64_t &count); 517 518 int GetUploadCountInner(const Timestamp ×tamp, SqliteQueryHelper &helper, std::string &sql, int64_t &count); 519 520 int LogicDeleteCloudData(const std::string &tableName, const VBucket &vBucket, 521 const TableSchema &tableSchema, const TrackerTable &trackerTable); 522 523 bool AbortGetDownloadAssetGidIfNeed(const TableSchema &tableSchema, const std::string &gid, bool abortWithLimit, 524 uint32_t &count); 525 526 static constexpr const char *CONSISTENT_FLAG = "0x20"; 527 static constexpr const char *UPDATE_FLAG_CLOUD = "flag = 0"; 528 static constexpr const char *UPDATE_FLAG_WAIT_COMPENSATED_SYNC = "flag = flag | 0x10"; 529 static constexpr const char *FLAG_IS_WAIT_COMPENSATED_SYNC = 530 "(a.flag & 0x10 != 0 and a.status = 0) or a.status = 1"; 531 static constexpr const char *FLAG_IS_WAIT_COMPENSATED_CONTAIN_DOWNLOAD_SYNC = 532 "(a.flag & 0x10 != 0 and a.status = 0) or a.status = 1 or (a.flag & 0x1000 != 0 and a.status = 0)"; 533 534 std::string baseTblName_; 535 TableInfo table_; // Always operating table, user table when get, device table when put. 536 TableSchema tableSchema_; // for cloud table 537 RelationalSchemaObject localSchema_; 538 539 DistributedTableMode mode_; 540 541 std::map<int32_t, std::function<int(int, const VBucket &, const Field &, sqlite3_stmt *)>> bindCloudFieldFuncMap_; 542 std::map<int32_t, std::function<int(const VBucket &, const Field &, std::vector<uint8_t> &)>> toVectorFuncMap_; 543 544 std::atomic<bool> isLogicDelete_; 545 std::shared_ptr<IAssetLoader> assetLoader_; 546 547 PutDataMode putDataMode_; 548 MarkFlagOption markFlagOption_; 549 550 std::atomic<int32_t> maxUploadCount_; 551 std::atomic<int32_t> maxUploadSize_; 552 }; 553 } // namespace DistributedDB 554 #endif 555 #endif // SQLITE_SINGLE_VER_RELATIONAL_STORAGE_EXECUTOR_H 556