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 #ifdef RELATIONAL_STORE
16 #include "relational_sync_able_storage.h"
17 
18 #include <utility>
19 
20 #include "cloud/cloud_db_constant.h"
21 #include "cloud/cloud_storage_utils.h"
22 #include "concurrent_adapter.h"
23 #include "data_compression.h"
24 #include "db_common.h"
25 #include "db_dfx_adapter.h"
26 #include "generic_single_ver_kv_entry.h"
27 #include "platform_specific.h"
28 #include "query_utils.h"
29 #include "relational_remote_query_continue_token.h"
30 #include "relational_sync_data_inserter.h"
31 #include "res_finalizer.h"
32 #include "runtime_context.h"
33 #include "time_helper.h"
34 
35 namespace DistributedDB {
36 namespace {
TriggerCloseAutoLaunchConn(const RelationalDBProperties & properties)37 void TriggerCloseAutoLaunchConn(const RelationalDBProperties &properties)
38 {
39     static constexpr const char *CLOSE_CONN_TASK = "auto launch close relational connection";
40     (void)RuntimeContext::GetInstance()->ScheduleQueuedTask(
41         std::string(CLOSE_CONN_TASK),
42         [properties] { RuntimeContext::GetInstance()->CloseAutoLaunchConnection(DBTypeInner::DB_RELATION, properties); }
43     );
44 }
45 }
46 
RelationalSyncAbleStorage(std::shared_ptr<SQLiteSingleRelationalStorageEngine> engine)47 RelationalSyncAbleStorage::RelationalSyncAbleStorage(std::shared_ptr<SQLiteSingleRelationalStorageEngine> engine)
48     : storageEngine_(std::move(engine)),
49       reusedHandle_(nullptr),
50       isCachedOption_(false)
51 {}
52 
~RelationalSyncAbleStorage()53 RelationalSyncAbleStorage::~RelationalSyncAbleStorage()
54 {
55     syncAbleEngine_ = nullptr;
56 }
57 
58 // Get interface type of this relational db.
GetInterfaceType() const59 int RelationalSyncAbleStorage::GetInterfaceType() const
60 {
61     return SYNC_RELATION;
62 }
63 
64 // Get the interface ref-count, in order to access asynchronously.
IncRefCount()65 void RelationalSyncAbleStorage::IncRefCount()
66 {
67     LOGD("RelationalSyncAbleStorage ref +1");
68     IncObjRef(this);
69 }
70 
71 // Drop the interface ref-count.
DecRefCount()72 void RelationalSyncAbleStorage::DecRefCount()
73 {
74     LOGD("RelationalSyncAbleStorage ref -1");
75     DecObjRef(this);
76 }
77 
78 // Get the identifier of this rdb.
GetIdentifier() const79 std::vector<uint8_t> RelationalSyncAbleStorage::GetIdentifier() const
80 {
81     std::string identifier = storageEngine_->GetIdentifier();
82     return std::vector<uint8_t>(identifier.begin(), identifier.end());
83 }
84 
GetDualTupleIdentifier() const85 std::vector<uint8_t> RelationalSyncAbleStorage::GetDualTupleIdentifier() const
86 {
87     std::string identifier = storageEngine_->GetProperties().GetStringProp(
88         DBProperties::DUAL_TUPLE_IDENTIFIER_DATA, "");
89     std::vector<uint8_t> identifierVect(identifier.begin(), identifier.end());
90     return identifierVect;
91 }
92 
93 // Get the max timestamp of all entries in database.
GetMaxTimestamp(Timestamp & timestamp) const94 void RelationalSyncAbleStorage::GetMaxTimestamp(Timestamp &timestamp) const
95 {
96     int errCode = E_OK;
97     auto handle = GetHandle(false, errCode, OperatePerm::NORMAL_PERM);
98     if (handle == nullptr) {
99         return;
100     }
101     timestamp = 0;
102     errCode = handle->GetMaxTimestamp(storageEngine_->GetSchema().GetTableNames(), timestamp);
103     if (errCode != E_OK) {
104         LOGE("GetMaxTimestamp failed, errCode:%d", errCode);
105         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
106     }
107     ReleaseHandle(handle);
108     return;
109 }
110 
GetMaxTimestamp(const std::string & tableName,Timestamp & timestamp) const111 int RelationalSyncAbleStorage::GetMaxTimestamp(const std::string &tableName, Timestamp &timestamp) const
112 {
113     int errCode = E_OK;
114     auto handle = GetHandle(false, errCode, OperatePerm::NORMAL_PERM);
115     if (handle == nullptr) {
116         return errCode;
117     }
118     timestamp = 0;
119     errCode = handle->GetMaxTimestamp({ tableName }, timestamp);
120     if (errCode != E_OK) {
121         LOGE("GetMaxTimestamp failed, errCode:%d", errCode);
122         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
123     }
124     ReleaseHandle(handle);
125     return errCode;
126 }
127 
GetHandle(bool isWrite,int & errCode,OperatePerm perm) const128 SQLiteSingleVerRelationalStorageExecutor *RelationalSyncAbleStorage::GetHandle(bool isWrite, int &errCode,
129     OperatePerm perm) const
130 {
131     if (storageEngine_ == nullptr) {
132         errCode = -E_INVALID_DB;
133         return nullptr;
134     }
135     auto handle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(
136         storageEngine_->FindExecutor(isWrite, perm, errCode));
137     if (handle == nullptr) {
138         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
139     }
140     return handle;
141 }
142 
GetHandleExpectTransaction(bool isWrite,int & errCode,OperatePerm perm) const143 SQLiteSingleVerRelationalStorageExecutor *RelationalSyncAbleStorage::GetHandleExpectTransaction(bool isWrite,
144     int &errCode, OperatePerm perm) const
145 {
146     if (storageEngine_ == nullptr) {
147         errCode = -E_INVALID_DB;
148         return nullptr;
149     }
150     if (transactionHandle_ != nullptr) {
151         return transactionHandle_;
152     }
153     auto handle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(
154         storageEngine_->FindExecutor(isWrite, perm, errCode));
155     if (errCode != E_OK) {
156         ReleaseHandle(handle);
157         handle = nullptr;
158     }
159     return handle;
160 }
161 
ReleaseHandle(SQLiteSingleVerRelationalStorageExecutor * & handle) const162 void RelationalSyncAbleStorage::ReleaseHandle(SQLiteSingleVerRelationalStorageExecutor *&handle) const
163 {
164     if (storageEngine_ == nullptr) {
165         return;
166     }
167     StorageExecutor *databaseHandle = handle;
168     storageEngine_->Recycle(databaseHandle);
169     std::function<void()> listener = nullptr;
170     {
171         std::lock_guard<std::mutex> autoLock(heartBeatMutex_);
172         listener = heartBeatListener_;
173     }
174     if (listener) {
175         listener();
176     }
177 }
178 
179 // Get meta data associated with the given key.
GetMetaData(const Key & key,Value & value) const180 int RelationalSyncAbleStorage::GetMetaData(const Key &key, Value &value) const
181 {
182     if (storageEngine_ == nullptr) {
183         return -E_INVALID_DB;
184     }
185     if (key.size() > DBConstant::MAX_KEY_SIZE) {
186         return -E_INVALID_ARGS;
187     }
188     int errCode = E_OK;
189     auto handle = GetHandle(false, errCode, OperatePerm::NORMAL_PERM);
190     if (handle == nullptr) {
191         return errCode;
192     }
193     errCode = handle->GetKvData(key, value);
194     if (errCode != E_OK && errCode != -E_NOT_FOUND) {
195         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
196     }
197     ReleaseHandle(handle);
198     return errCode;
199 }
200 
201 // Put meta data as a key-value entry.
PutMetaData(const Key & key,const Value & value)202 int RelationalSyncAbleStorage::PutMetaData(const Key &key, const Value &value)
203 {
204     if (storageEngine_ == nullptr) {
205         return -E_INVALID_DB;
206     }
207     int errCode = E_OK;
208     auto *handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
209     if (handle == nullptr) {
210         return errCode;
211     }
212 
213     errCode = handle->PutKvData(key, value); // meta doesn't need time.
214     if (errCode != E_OK) {
215         LOGE("Put kv data err:%d", errCode);
216         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
217     }
218     ReleaseHandle(handle);
219     return errCode;
220 }
221 
PutMetaData(const Key & key,const Value & value,bool isInTransaction)222 int RelationalSyncAbleStorage::PutMetaData(const Key &key, const Value &value, bool isInTransaction)
223 {
224     if (storageEngine_ == nullptr) {
225         return -E_INVALID_DB;
226     }
227     int errCode = E_OK;
228     SQLiteSingleVerRelationalStorageExecutor *handle = nullptr;
229     std::unique_lock<std::mutex> handLock(reusedHandleMutex_, std::defer_lock);
230 
231     // try to recycle using the handle
232     if (isInTransaction) {
233         handLock.lock();
234         if (reusedHandle_ != nullptr) {
235             handle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(reusedHandle_);
236         } else {
237             isInTransaction = false;
238             handLock.unlock();
239         }
240     }
241 
242     if (handle == nullptr) {
243         handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
244         if (handle == nullptr) {
245             return errCode;
246         }
247     }
248 
249     errCode = handle->PutKvData(key, value);
250     if (errCode != E_OK) {
251         LOGE("Put kv data err:%d", errCode);
252         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
253     }
254     if (!isInTransaction) {
255         ReleaseHandle(handle);
256     }
257     return errCode;
258 }
259 
260 // Delete multiple meta data records in a transaction.
DeleteMetaData(const std::vector<Key> & keys)261 int RelationalSyncAbleStorage::DeleteMetaData(const std::vector<Key> &keys)
262 {
263     if (storageEngine_ == nullptr) {
264         return -E_INVALID_DB;
265     }
266     for (const auto &key : keys) {
267         if (key.empty() || key.size() > DBConstant::MAX_KEY_SIZE) {
268             return -E_INVALID_ARGS;
269         }
270     }
271     int errCode = E_OK;
272     auto handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
273     if (handle == nullptr) {
274         return errCode;
275     }
276 
277     handle->StartTransaction(TransactType::IMMEDIATE);
278     errCode = handle->DeleteMetaData(keys);
279     if (errCode != E_OK) {
280         handle->Rollback();
281         LOGE("[SinStore] DeleteMetaData failed, errCode = %d", errCode);
282         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
283     } else {
284         handle->Commit();
285     }
286     ReleaseHandle(handle);
287     return errCode;
288 }
289 
290 // Delete multiple meta data records with key prefix in a transaction.
DeleteMetaDataByPrefixKey(const Key & keyPrefix) const291 int RelationalSyncAbleStorage::DeleteMetaDataByPrefixKey(const Key &keyPrefix) const
292 {
293     if (storageEngine_ == nullptr) {
294         return -E_INVALID_DB;
295     }
296     if (keyPrefix.empty() || keyPrefix.size() > DBConstant::MAX_KEY_SIZE) {
297         return -E_INVALID_ARGS;
298     }
299 
300     int errCode = E_OK;
301     auto handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
302     if (handle == nullptr) {
303         return errCode;
304     }
305 
306     errCode = handle->DeleteMetaDataByPrefixKey(keyPrefix);
307     if (errCode != E_OK) {
308         LOGE("[SinStore] DeleteMetaData by prefix key failed, errCode = %d", errCode);
309         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
310     }
311     ReleaseHandle(handle);
312     return errCode;
313 }
314 
315 // Get all meta data keys.
GetAllMetaKeys(std::vector<Key> & keys) const316 int RelationalSyncAbleStorage::GetAllMetaKeys(std::vector<Key> &keys) const
317 {
318     if (storageEngine_ == nullptr) {
319         return -E_INVALID_DB;
320     }
321     int errCode = E_OK;
322     auto *handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
323     if (handle == nullptr) {
324         return errCode;
325     }
326 
327     errCode = handle->GetAllMetaKeys(keys);
328     if (errCode != E_OK) {
329         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
330     }
331     ReleaseHandle(handle);
332     return errCode;
333 }
334 
GetDbProperties() const335 const RelationalDBProperties &RelationalSyncAbleStorage::GetDbProperties() const
336 {
337     return storageEngine_->GetProperties();
338 }
339 
GetKvEntriesByDataItems(std::vector<SingleVerKvEntry * > & entries,std::vector<DataItem> & dataItems)340 static int GetKvEntriesByDataItems(std::vector<SingleVerKvEntry *> &entries, std::vector<DataItem> &dataItems)
341 {
342     int errCode = E_OK;
343     for (auto &item : dataItems) {
344         auto entry = new (std::nothrow) GenericSingleVerKvEntry();
345         if (entry == nullptr) {
346             errCode = -E_OUT_OF_MEMORY;
347             LOGE("GetKvEntries failed, errCode:%d", errCode);
348             SingleVerKvEntry::Release(entries);
349             break;
350         }
351         entry->SetEntryData(std::move(item));
352         entries.push_back(entry);
353     }
354     return errCode;
355 }
356 
GetDataItemSerialSize(const DataItem & item,size_t appendLen)357 static size_t GetDataItemSerialSize(const DataItem &item, size_t appendLen)
358 {
359     // timestamp and local flag: 3 * uint64_t, version(uint32_t), key, value, origin dev and the padding size.
360     // the size would not be very large.
361     static const size_t maxOrigDevLength = 40;
362     size_t devLength = std::max(maxOrigDevLength, item.origDev.size());
363     size_t dataSize = (Parcel::GetUInt64Len() * 3 + Parcel::GetUInt32Len() + Parcel::GetVectorCharLen(item.key) +
364                        Parcel::GetVectorCharLen(item.value) + devLength + appendLen);
365     return dataSize;
366 }
367 
CanHoldDeletedData(const std::vector<DataItem> & dataItems,const DataSizeSpecInfo & dataSizeInfo,size_t appendLen)368 static bool CanHoldDeletedData(const std::vector<DataItem> &dataItems, const DataSizeSpecInfo &dataSizeInfo,
369     size_t appendLen)
370 {
371     bool reachThreshold = (dataItems.size() >= dataSizeInfo.packetSize);
372     for (size_t i = 0, blockSize = 0; !reachThreshold && i < dataItems.size(); i++) {
373         blockSize += GetDataItemSerialSize(dataItems[i], appendLen);
374         reachThreshold = (blockSize >= dataSizeInfo.blockSize * DBConstant::QUERY_SYNC_THRESHOLD);
375     }
376     return !reachThreshold;
377 }
378 
ProcessContinueTokenForQuerySync(const std::vector<DataItem> & dataItems,int & errCode,SQLiteSingleVerRelationalContinueToken * & token)379 static void ProcessContinueTokenForQuerySync(const std::vector<DataItem> &dataItems, int &errCode,
380     SQLiteSingleVerRelationalContinueToken *&token)
381 {
382     if (errCode != -E_UNFINISHED) { // Error happened or get data finished. Token should be cleared.
383         delete token;
384         token = nullptr;
385         return;
386     }
387 
388     if (dataItems.empty()) {
389         errCode = -E_INTERNAL_ERROR;
390         LOGE("Get data unfinished but data items is empty.");
391         delete token;
392         token = nullptr;
393         return;
394     }
395     token->SetNextBeginTime(dataItems.back());
396     token->UpdateNextSyncOffset(dataItems.size());
397 }
398 
399 /**
400  * Caller must ensure that parameter token is valid.
401  * If error happened, token will be deleted here.
402  */
GetSyncDataForQuerySync(std::vector<DataItem> & dataItems,SQLiteSingleVerRelationalContinueToken * & token,const DataSizeSpecInfo & dataSizeInfo) const403 int RelationalSyncAbleStorage::GetSyncDataForQuerySync(std::vector<DataItem> &dataItems,
404     SQLiteSingleVerRelationalContinueToken *&token, const DataSizeSpecInfo &dataSizeInfo) const
405 {
406     if (storageEngine_ == nullptr) {
407         return -E_INVALID_DB;
408     }
409 
410     int errCode = E_OK;
411     auto handle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(storageEngine_->FindExecutor(false,
412         OperatePerm::NORMAL_PERM, errCode));
413     if (handle == nullptr) {
414         goto ERROR;
415     }
416 
417     do {
418         errCode = handle->GetSyncDataByQuery(dataItems,
419             Parcel::GetAppendedLen(),
420             dataSizeInfo,
421             [token](sqlite3 *db, sqlite3_stmt *&queryStmt, sqlite3_stmt *&fullStmt, bool &isGettingDeletedData) {
422                 return token->GetStatement(db, queryStmt, fullStmt, isGettingDeletedData);
423             }, storageEngine_->GetSchema().GetTable(token->GetQuery().GetTableName()));
424         if (errCode == -E_FINISHED) {
425             token->FinishGetData();
426             errCode = token->IsGetAllDataFinished() ? E_OK : -E_UNFINISHED;
427         }
428     } while (errCode == -E_UNFINISHED && CanHoldDeletedData(dataItems, dataSizeInfo, Parcel::GetAppendedLen()));
429 
430 ERROR:
431     if (errCode != -E_UNFINISHED && errCode != E_OK) { // Error happened.
432         dataItems.clear();
433     }
434     ProcessContinueTokenForQuerySync(dataItems, errCode, token);
435     ReleaseHandle(handle);
436     return errCode;
437 }
438 
439 // use kv struct data to sync
440 // Get the data which would be synced with query condition
GetSyncData(QueryObject & query,const SyncTimeRange & timeRange,const DataSizeSpecInfo & dataSizeInfo,ContinueToken & continueStmtToken,std::vector<SingleVerKvEntry * > & entries) const441 int RelationalSyncAbleStorage::GetSyncData(QueryObject &query, const SyncTimeRange &timeRange,
442     const DataSizeSpecInfo &dataSizeInfo, ContinueToken &continueStmtToken,
443     std::vector<SingleVerKvEntry *> &entries) const
444 {
445     if (!timeRange.IsValid()) {
446         return -E_INVALID_ARGS;
447     }
448     query.SetSchema(storageEngine_->GetSchema());
449     auto token = new (std::nothrow) SQLiteSingleVerRelationalContinueToken(timeRange, query);
450     if (token == nullptr) {
451         LOGE("[SingleVerNStore] Allocate continue token failed.");
452         return -E_OUT_OF_MEMORY;
453     }
454 
455     continueStmtToken = static_cast<ContinueToken>(token);
456     return GetSyncDataNext(entries, continueStmtToken, dataSizeInfo);
457 }
458 
GetSyncDataNext(std::vector<SingleVerKvEntry * > & entries,ContinueToken & continueStmtToken,const DataSizeSpecInfo & dataSizeInfo) const459 int RelationalSyncAbleStorage::GetSyncDataNext(std::vector<SingleVerKvEntry *> &entries,
460     ContinueToken &continueStmtToken, const DataSizeSpecInfo &dataSizeInfo) const
461 {
462     auto token = static_cast<SQLiteSingleVerRelationalContinueToken *>(continueStmtToken);
463     if (!token->CheckValid()) {
464         return -E_INVALID_ARGS;
465     }
466     RelationalSchemaObject schema = storageEngine_->GetSchema();
467     const auto fieldInfos = schema.GetTable(token->GetQuery().GetTableName()).GetFieldInfos();
468     std::vector<std::string> fieldNames;
469     fieldNames.reserve(fieldInfos.size());
470     for (const auto &fieldInfo : fieldInfos) { // order by cid
471         fieldNames.push_back(fieldInfo.GetFieldName());
472     }
473     token->SetFieldNames(fieldNames);
474 
475     std::vector<DataItem> dataItems;
476     int errCode = GetSyncDataForQuerySync(dataItems, token, dataSizeInfo);
477     if (errCode != E_OK && errCode != -E_UNFINISHED) { // The code need be sent to outside except new error happened.
478         continueStmtToken = static_cast<ContinueToken>(token);
479         return errCode;
480     }
481 
482     int innerCode = GetKvEntriesByDataItems(entries, dataItems);
483     if (innerCode != E_OK) {
484         errCode = innerCode;
485         delete token;
486         token = nullptr;
487     }
488     continueStmtToken = static_cast<ContinueToken>(token);
489     return errCode;
490 }
491 
492 namespace {
ConvertEntries(std::vector<SingleVerKvEntry * > entries)493 std::vector<DataItem> ConvertEntries(std::vector<SingleVerKvEntry *> entries)
494 {
495     std::vector<DataItem> dataItems;
496     for (const auto &itemEntry : entries) {
497         GenericSingleVerKvEntry *entry = static_cast<GenericSingleVerKvEntry *>(itemEntry);
498         if (entry != nullptr) {
499             DataItem item;
500             item.origDev = entry->GetOrigDevice();
501             item.flag = entry->GetFlag();
502             item.timestamp = entry->GetTimestamp();
503             item.writeTimestamp = entry->GetWriteTimestamp();
504             entry->GetKey(item.key);
505             entry->GetValue(item.value);
506             entry->GetHashKey(item.hashKey);
507             dataItems.push_back(item);
508         }
509     }
510     return dataItems;
511 }
512 }
513 
PutSyncDataWithQuery(const QueryObject & object,const std::vector<SingleVerKvEntry * > & entries,const DeviceID & deviceName)514 int RelationalSyncAbleStorage::PutSyncDataWithQuery(const QueryObject &object,
515     const std::vector<SingleVerKvEntry *> &entries, const DeviceID &deviceName)
516 {
517     std::vector<DataItem> dataItems = ConvertEntries(entries);
518     return PutSyncData(object, dataItems, deviceName);
519 }
520 
521 namespace {
GetCollaborationMode(const std::shared_ptr<SQLiteSingleRelationalStorageEngine> & engine)522 inline DistributedTableMode GetCollaborationMode(const std::shared_ptr<SQLiteSingleRelationalStorageEngine> &engine)
523 {
524     return static_cast<DistributedTableMode>(engine->GetProperties().GetIntProp(
525         RelationalDBProperties::DISTRIBUTED_TABLE_MODE, DistributedTableMode::SPLIT_BY_DEVICE));
526 }
527 
IsCollaborationMode(const std::shared_ptr<SQLiteSingleRelationalStorageEngine> & engine)528 inline bool IsCollaborationMode(const std::shared_ptr<SQLiteSingleRelationalStorageEngine> &engine)
529 {
530     return GetCollaborationMode(engine) == DistributedTableMode::COLLABORATION;
531 }
532 }
533 
SaveSyncDataItems(const QueryObject & object,std::vector<DataItem> & dataItems,const std::string & deviceName)534 int RelationalSyncAbleStorage::SaveSyncDataItems(const QueryObject &object, std::vector<DataItem> &dataItems,
535     const std::string &deviceName)
536 {
537     int errCode = E_OK;
538     LOGD("[RelationalSyncAbleStorage::SaveSyncDataItems] Get write handle.");
539     QueryObject query = object;
540     query.SetSchema(storageEngine_->GetSchema());
541 
542     RelationalSchemaObject remoteSchema;
543     errCode = GetRemoteDeviceSchema(deviceName, remoteSchema);
544     if (errCode != E_OK) {
545         LOGE("Find remote schema failed. err=%d", errCode);
546         return errCode;
547     }
548 
549     StoreInfo info = GetStoreInfo();
550     auto inserter = RelationalSyncDataInserter::CreateInserter(deviceName, query, storageEngine_->GetSchema(),
551         remoteSchema.GetTable(query.GetTableName()).GetFieldInfos(), info);
552     inserter.SetEntries(dataItems);
553 
554     auto *handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
555     if (handle == nullptr) {
556         return errCode;
557     }
558 
559     DBDfxAdapter::StartTracing();
560 
561     errCode = handle->SaveSyncItems(inserter);
562 
563     DBDfxAdapter::FinishTracing();
564     if (errCode == E_OK) {
565         // dataItems size > 0 now because already check before
566         // all dataItems will write into db now, so need to observer notify here
567         // if some dataItems will not write into db in the future, observer notify here need change
568         ChangedData data;
569         TriggerObserverAction(deviceName, std::move(data), false);
570     }
571 
572     ReleaseHandle(handle);
573     return errCode;
574 }
575 
PutSyncData(const QueryObject & query,std::vector<DataItem> & dataItems,const std::string & deviceName)576 int RelationalSyncAbleStorage::PutSyncData(const QueryObject &query, std::vector<DataItem> &dataItems,
577     const std::string &deviceName)
578 {
579     if (deviceName.length() > DBConstant::MAX_DEV_LENGTH) {
580         LOGW("Device length is invalid for sync put");
581         return -E_INVALID_ARGS;
582     }
583 
584     int errCode = SaveSyncDataItems(query, dataItems, deviceName); // Currently true to check value content
585     if (errCode != E_OK) {
586         LOGE("[Relational] PutSyncData errCode:%d", errCode);
587         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
588     }
589     return errCode;
590 }
591 
RemoveDeviceData(const std::string & deviceName,bool isNeedNotify)592 int RelationalSyncAbleStorage::RemoveDeviceData(const std::string &deviceName, bool isNeedNotify)
593 {
594     (void) deviceName;
595     (void) isNeedNotify;
596     return -E_NOT_SUPPORT;
597 }
598 
GetSchemaInfo() const599 RelationalSchemaObject RelationalSyncAbleStorage::GetSchemaInfo() const
600 {
601     return storageEngine_->GetSchema();
602 }
603 
GetSecurityOption(SecurityOption & option) const604 int RelationalSyncAbleStorage::GetSecurityOption(SecurityOption &option) const
605 {
606     std::lock_guard<std::mutex> autoLock(securityOptionMutex_);
607     if (isCachedOption_) {
608         option = securityOption_;
609         return E_OK;
610     }
611     std::string dbPath = storageEngine_->GetProperties().GetStringProp(DBProperties::DATA_DIR, "");
612     int errCode = RuntimeContext::GetInstance()->GetSecurityOption(dbPath, securityOption_);
613     if (errCode == E_OK) {
614         option = securityOption_;
615         isCachedOption_ = true;
616     }
617     return errCode;
618 }
619 
NotifyRemotePushFinished(const std::string & deviceId) const620 void RelationalSyncAbleStorage::NotifyRemotePushFinished(const std::string &deviceId) const
621 {
622     return;
623 }
624 
625 // Get the timestamp when database created or imported
GetDatabaseCreateTimestamp(Timestamp & outTime) const626 int RelationalSyncAbleStorage::GetDatabaseCreateTimestamp(Timestamp &outTime) const
627 {
628     return OS::GetCurrentSysTimeInMicrosecond(outTime);
629 }
630 
GetTablesQuery()631 std::vector<QuerySyncObject> RelationalSyncAbleStorage::GetTablesQuery()
632 {
633     auto tableNames = storageEngine_->GetSchema().GetTableNames();
634     std::vector<QuerySyncObject> queries;
635     queries.reserve(tableNames.size());
636     for (const auto &it : tableNames) {
637         queries.emplace_back(Query::Select(it));
638     }
639     return queries;
640 }
641 
LocalDataChanged(int notifyEvent,std::vector<QuerySyncObject> & queryObj)642 int RelationalSyncAbleStorage::LocalDataChanged(int notifyEvent, std::vector<QuerySyncObject> &queryObj)
643 {
644     (void) queryObj;
645     return -E_NOT_SUPPORT;
646 }
647 
InterceptData(std::vector<SingleVerKvEntry * > & entries,const std::string & sourceID,const std::string & targetID,bool isPush) const648 int RelationalSyncAbleStorage::InterceptData(std::vector<SingleVerKvEntry *> &entries, const std::string &sourceID,
649     const std::string &targetID, bool isPush) const
650 {
651     return E_OK;
652 }
653 
CreateDistributedDeviceTable(const std::string & device,const RelationalSyncStrategy & syncStrategy)654 int RelationalSyncAbleStorage::CreateDistributedDeviceTable(const std::string &device,
655     const RelationalSyncStrategy &syncStrategy)
656 {
657     auto mode = storageEngine_->GetProperties().GetIntProp(RelationalDBProperties::DISTRIBUTED_TABLE_MODE,
658         DistributedTableMode::SPLIT_BY_DEVICE);
659     if (mode != DistributedTableMode::SPLIT_BY_DEVICE) {
660         LOGD("No need create device table in COLLABORATION mode.");
661         return E_OK;
662     }
663 
664     int errCode = E_OK;
665     auto *handle = GetHandle(true, errCode, OperatePerm::NORMAL_PERM);
666     if (handle == nullptr) {
667         return errCode;
668     }
669 
670     errCode = handle->StartTransaction(TransactType::IMMEDIATE);
671     if (errCode != E_OK) {
672         LOGE("Start transaction failed:%d", errCode);
673         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
674         ReleaseHandle(handle);
675         return errCode;
676     }
677 
678     StoreInfo info = GetStoreInfo();
679     for (const auto &[table, strategy] : syncStrategy) {
680         if (!strategy.permitSync) {
681             continue;
682         }
683 
684         errCode = handle->CreateDistributedDeviceTable(device, storageEngine_->GetSchema().GetTable(table), info);
685         if (errCode != E_OK) {
686             LOGE("Create distributed device table failed. %d", errCode);
687             break;
688         }
689     }
690 
691     if (errCode == E_OK) {
692         errCode = handle->Commit();
693     } else {
694         (void)handle->Rollback();
695     }
696 
697     ReleaseHandle(handle);
698     return errCode;
699 }
700 
RegisterSchemaChangedCallback(const std::function<void ()> & callback)701 int RelationalSyncAbleStorage::RegisterSchemaChangedCallback(const std::function<void()> &callback)
702 {
703     std::lock_guard lock(onSchemaChangedMutex_);
704     onSchemaChanged_ = callback;
705     return E_OK;
706 }
707 
NotifySchemaChanged()708 void RelationalSyncAbleStorage::NotifySchemaChanged()
709 {
710     std::lock_guard lock(onSchemaChangedMutex_);
711     if (onSchemaChanged_) {
712         LOGD("Notify relational schema was changed");
713         onSchemaChanged_();
714     }
715 }
GetCompressionAlgo(std::set<CompressAlgorithm> & algorithmSet) const716 int RelationalSyncAbleStorage::GetCompressionAlgo(std::set<CompressAlgorithm> &algorithmSet) const
717 {
718     algorithmSet.clear();
719     DataCompression::GetCompressionAlgo(algorithmSet);
720     return E_OK;
721 }
722 
RegisterObserverAction(uint64_t connectionId,const StoreObserver * observer,const RelationalObserverAction & action)723 int RelationalSyncAbleStorage::RegisterObserverAction(uint64_t connectionId, const StoreObserver *observer,
724     const RelationalObserverAction &action)
725 {
726     int errCode = E_OK;
727     TaskHandle handle = ConcurrentAdapter::ScheduleTaskH([this, connectionId, observer, action, &errCode] () mutable {
728         ADAPTER_AUTO_LOCK(lock, dataChangeDeviceMutex_);
729         auto it = dataChangeCallbackMap_.find(connectionId);
730         if (it != dataChangeCallbackMap_.end()) {
731             if (it->second.find(observer) != it->second.end()) {
732                 LOGE("obsever already registered");
733                 errCode = -E_ALREADY_SET;
734                 return;
735             }
736             if (it->second.size() >= DBConstant::MAX_OBSERVER_COUNT) {
737                 LOGE("The number of relational observers has been over limit");
738                 errCode = -E_MAX_LIMITS;
739                 return;
740             }
741             it->second[observer] = action;
742         } else {
743             dataChangeCallbackMap_[connectionId][observer] = action;
744         }
745         LOGI("register relational observer ok");
746     }, nullptr, &dataChangeCallbackMap_);
747     ADAPTER_WAIT(handle);
748     return errCode;
749 }
750 
UnRegisterObserverAction(uint64_t connectionId,const StoreObserver * observer)751 int RelationalSyncAbleStorage::UnRegisterObserverAction(uint64_t connectionId, const StoreObserver *observer)
752 {
753     if (observer == nullptr) {
754         EraseDataChangeCallback(connectionId);
755         return E_OK;
756     }
757     int errCode = -E_NOT_FOUND;
758     TaskHandle handle = ConcurrentAdapter::ScheduleTaskH([this, connectionId, observer, &errCode] () mutable {
759         ADAPTER_AUTO_LOCK(lock, dataChangeDeviceMutex_);
760         auto it = dataChangeCallbackMap_.find(connectionId);
761         if (it == dataChangeCallbackMap_.end()) {
762             return;
763         }
764         auto action = it->second.find(observer);
765         if (action == it->second.end()) {
766             return;
767         }
768         it->second.erase(action);
769         LOGI("unregister relational observer.");
770         if (it->second.empty()) {
771             dataChangeCallbackMap_.erase(it);
772             LOGI("observer for this delegate is zero now");
773         }
774         errCode = E_OK;
775     }, nullptr, &dataChangeCallbackMap_);
776     ADAPTER_WAIT(handle);
777     return errCode;
778 }
779 
ExecuteDataChangeCallback(const std::pair<uint64_t,std::map<const StoreObserver *,RelationalObserverAction>> & item,const std::string & deviceName,const ChangedData & changedData,bool isChangedData,int & observerCnt)780 void RelationalSyncAbleStorage::ExecuteDataChangeCallback(
781     const std::pair<uint64_t, std::map<const StoreObserver *, RelationalObserverAction>> &item,
782     const std::string &deviceName, const ChangedData &changedData, bool isChangedData, int &observerCnt)
783 {
784     for (auto &action : item.second) {
785         if (action.second == nullptr) {
786             continue;
787         }
788         observerCnt++;
789         ChangedData observerChangeData = changedData;
790         if (action.first != nullptr) {
791             FilterChangeDataByDetailsType(observerChangeData, action.first->GetCallbackDetailsType());
792         }
793         action.second(deviceName, std::move(observerChangeData), isChangedData);
794     }
795 }
796 
TriggerObserverAction(const std::string & deviceName,ChangedData && changedData,bool isChangedData)797 void RelationalSyncAbleStorage::TriggerObserverAction(const std::string &deviceName,
798     ChangedData &&changedData, bool isChangedData)
799 {
800     IncObjRef(this);
801     int taskErrCode =
802         ConcurrentAdapter::ScheduleTask([this, deviceName, changedData, isChangedData] () mutable {
803             LOGD("begin to trigger relational observer.");
804             int observerCnt = 0;
805             ADAPTER_AUTO_LOCK(lock, dataChangeDeviceMutex_);
806             for (const auto &item : dataChangeCallbackMap_) {
807                 ExecuteDataChangeCallback(item, deviceName, changedData, isChangedData, observerCnt);
808             }
809             LOGD("relational observer size = %d", observerCnt);
810             DecObjRef(this);
811         }, &dataChangeCallbackMap_);
812     if (taskErrCode != E_OK) {
813         LOGE("TriggerObserverAction scheduletask retCode=%d", taskErrCode);
814         DecObjRef(this);
815     }
816 }
817 
RegisterHeartBeatListener(const std::function<void ()> & listener)818 void RelationalSyncAbleStorage::RegisterHeartBeatListener(const std::function<void()> &listener)
819 {
820     std::lock_guard<std::mutex> autoLock(heartBeatMutex_);
821     heartBeatListener_ = listener;
822 }
823 
CheckAndInitQueryCondition(QueryObject & query) const824 int RelationalSyncAbleStorage::CheckAndInitQueryCondition(QueryObject &query) const
825 {
826     RelationalSchemaObject schema = storageEngine_->GetSchema();
827     TableInfo table = schema.GetTable(query.GetTableName());
828     if (!table.IsValid()) {
829         LOGE("Query table is not a distributed table.");
830         return -E_DISTRIBUTED_SCHEMA_NOT_FOUND;
831     }
832     if (table.GetTableSyncType() == CLOUD_COOPERATION) {
833         LOGE("cloud table mode is not support");
834         return -E_NOT_SUPPORT;
835     }
836     query.SetSchema(schema);
837 
838     int errCode = E_OK;
839     auto *handle = GetHandle(true, errCode);
840     if (handle == nullptr) {
841         return errCode;
842     }
843 
844     errCode = handle->CheckQueryObjectLegal(table, query, schema.GetSchemaVersion());
845     if (errCode != E_OK) {
846         LOGE("Check relational query condition failed. %d", errCode);
847         TriggerCloseAutoLaunchConn(storageEngine_->GetProperties());
848     }
849 
850     ReleaseHandle(handle);
851     return errCode;
852 }
853 
CheckCompatible(const std::string & schema,uint8_t type) const854 bool RelationalSyncAbleStorage::CheckCompatible(const std::string &schema, uint8_t type) const
855 {
856     // return true if is relational schema.
857     return !schema.empty() && ReadSchemaType(type) == SchemaType::RELATIVE;
858 }
859 
GetRemoteQueryData(const PreparedStmt & prepStmt,size_t packetSize,std::vector<std::string> & colNames,std::vector<RelationalRowData * > & data) const860 int RelationalSyncAbleStorage::GetRemoteQueryData(const PreparedStmt &prepStmt, size_t packetSize,
861     std::vector<std::string> &colNames, std::vector<RelationalRowData *> &data) const
862 {
863     if (IsCollaborationMode(storageEngine_) || !storageEngine_->GetSchema().IsSchemaValid()) {
864         return -E_NOT_SUPPORT;
865     }
866     if (prepStmt.GetOpCode() != PreparedStmt::ExecutorOperation::QUERY || !prepStmt.IsValid()) {
867         LOGE("[ExecuteQuery] invalid args");
868         return -E_INVALID_ARGS;
869     }
870     int errCode = E_OK;
871     auto handle = GetHandle(false, errCode, OperatePerm::NORMAL_PERM);
872     if (handle == nullptr) {
873         LOGE("[ExecuteQuery] get handle fail:%d", errCode);
874         return errCode;
875     }
876     errCode = handle->ExecuteQueryBySqlStmt(prepStmt.GetSql(), prepStmt.GetBindArgs(), packetSize, colNames, data);
877     if (errCode != E_OK) {
878         LOGE("[ExecuteQuery] ExecuteQueryBySqlStmt failed:%d", errCode);
879     }
880     ReleaseHandle(handle);
881     return errCode;
882 }
883 
ExecuteQuery(const PreparedStmt & prepStmt,size_t packetSize,RelationalRowDataSet & dataSet,ContinueToken & token) const884 int RelationalSyncAbleStorage::ExecuteQuery(const PreparedStmt &prepStmt, size_t packetSize,
885     RelationalRowDataSet &dataSet, ContinueToken &token) const
886 {
887     dataSet.Clear();
888     if (token == nullptr) {
889         // start query
890         std::vector<std::string> colNames;
891         std::vector<RelationalRowData *> data;
892         ResFinalizer finalizer([&data] { RelationalRowData::Release(data); });
893 
894         int errCode = GetRemoteQueryData(prepStmt, packetSize, colNames, data);
895         if (errCode != E_OK) {
896             return errCode;
897         }
898 
899         // create one token
900         token = static_cast<ContinueToken>(
901             new (std::nothrow) RelationalRemoteQueryContinueToken(std::move(colNames), std::move(data)));
902         if (token == nullptr) {
903             LOGE("ExecuteQuery OOM");
904             return -E_OUT_OF_MEMORY;
905         }
906     }
907 
908     auto remoteToken = static_cast<RelationalRemoteQueryContinueToken *>(token);
909     if (!remoteToken->CheckValid()) {
910         LOGE("ExecuteQuery invalid token");
911         return -E_INVALID_ARGS;
912     }
913 
914     int errCode = remoteToken->GetData(packetSize, dataSet);
915     if (errCode == -E_UNFINISHED) {
916         errCode = E_OK;
917     } else {
918         if (errCode != E_OK) {
919             dataSet.Clear();
920         }
921         delete remoteToken;
922         remoteToken = nullptr;
923         token = nullptr;
924     }
925     LOGI("ExecuteQuery finished, errCode:%d, size:%d", errCode, dataSet.GetSize());
926     return errCode;
927 }
928 
SaveRemoteDeviceSchema(const std::string & deviceId,const std::string & remoteSchema,uint8_t type)929 int RelationalSyncAbleStorage::SaveRemoteDeviceSchema(const std::string &deviceId, const std::string &remoteSchema,
930     uint8_t type)
931 {
932     if (ReadSchemaType(type) != SchemaType::RELATIVE) {
933         return -E_INVALID_ARGS;
934     }
935 
936     RelationalSchemaObject schemaObj;
937     int errCode = schemaObj.ParseFromSchemaString(remoteSchema);
938     if (errCode != E_OK) {
939         LOGE("Parse remote schema failed. err=%d", errCode);
940         return errCode;
941     }
942 
943     std::string keyStr = DBConstant::REMOTE_DEVICE_SCHEMA_KEY_PREFIX + DBCommon::TransferHashString(deviceId);
944     Key remoteSchemaKey(keyStr.begin(), keyStr.end());
945     Value remoteSchemaBuff(remoteSchema.begin(), remoteSchema.end());
946     errCode = PutMetaData(remoteSchemaKey, remoteSchemaBuff);
947     if (errCode != E_OK) {
948         LOGE("Save remote schema failed. err=%d", errCode);
949         return errCode;
950     }
951 
952     return remoteDeviceSchema_.Put(deviceId, remoteSchema);
953 }
954 
GetRemoteDeviceSchema(const std::string & deviceId,RelationalSchemaObject & schemaObj)955 int RelationalSyncAbleStorage::GetRemoteDeviceSchema(const std::string &deviceId, RelationalSchemaObject &schemaObj)
956 {
957     if (schemaObj.IsSchemaValid()) {
958         return -E_INVALID_ARGS;
959     }
960 
961     std::string remoteSchema;
962     int errCode = remoteDeviceSchema_.Get(deviceId, remoteSchema);
963     if (errCode == -E_NOT_FOUND) {
964         LOGW("Get remote device schema miss cached.");
965         std::string keyStr = DBConstant::REMOTE_DEVICE_SCHEMA_KEY_PREFIX + DBCommon::TransferHashString(deviceId);
966         Key remoteSchemaKey(keyStr.begin(), keyStr.end());
967         Value remoteSchemaBuff;
968         errCode = GetMetaData(remoteSchemaKey, remoteSchemaBuff);
969         if (errCode != E_OK) {
970             LOGE("Get remote device schema from meta failed. err=%d", errCode);
971             return errCode;
972         }
973         remoteSchema = std::string(remoteSchemaBuff.begin(), remoteSchemaBuff.end());
974         errCode = remoteDeviceSchema_.Put(deviceId, remoteSchema);
975     }
976 
977     if (errCode != E_OK) {
978         LOGE("Get remote device schema failed. err=%d", errCode);
979         return errCode;
980     }
981 
982     errCode = schemaObj.ParseFromSchemaString(remoteSchema);
983     if (errCode != E_OK) {
984         LOGE("Parse remote schema failed. err=%d", errCode);
985     }
986     return errCode;
987 }
988 
SetReusedHandle(StorageExecutor * handle)989 void RelationalSyncAbleStorage::SetReusedHandle(StorageExecutor *handle)
990 {
991     std::lock_guard<std::mutex> autoLock(reusedHandleMutex_);
992     reusedHandle_ = handle;
993 }
994 
ReleaseRemoteQueryContinueToken(ContinueToken & token) const995 void RelationalSyncAbleStorage::ReleaseRemoteQueryContinueToken(ContinueToken &token) const
996 {
997     auto remoteToken = static_cast<RelationalRemoteQueryContinueToken *>(token);
998     delete remoteToken;
999     remoteToken = nullptr;
1000     token = nullptr;
1001 }
1002 
GetStoreInfo() const1003 StoreInfo RelationalSyncAbleStorage::GetStoreInfo() const
1004 {
1005     StoreInfo info = {
1006         storageEngine_->GetProperties().GetStringProp(DBProperties::USER_ID, ""),
1007         storageEngine_->GetProperties().GetStringProp(DBProperties::APP_ID, ""),
1008         storageEngine_->GetProperties().GetStringProp(DBProperties::STORE_ID, "")
1009     };
1010     return info;
1011 }
1012 
StartTransaction(TransactType type)1013 int RelationalSyncAbleStorage::StartTransaction(TransactType type)
1014 {
1015     if (storageEngine_ == nullptr) {
1016         return -E_INVALID_DB;
1017     }
1018     std::unique_lock<std::shared_mutex> lock(transactionMutex_);
1019     if (transactionHandle_ != nullptr) {
1020         LOGD("Transaction started already.");
1021         return -E_TRANSACT_STATE;
1022     }
1023     int errCode = E_OK;
1024     auto *handle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(
1025         storageEngine_->FindExecutor(type == TransactType::IMMEDIATE, OperatePerm::NORMAL_PERM, errCode));
1026     if (handle == nullptr) {
1027         ReleaseHandle(handle);
1028         return errCode;
1029     }
1030     errCode = handle->StartTransaction(type);
1031     if (errCode != E_OK) {
1032         ReleaseHandle(handle);
1033         return errCode;
1034     }
1035     transactionHandle_ = handle;
1036     return errCode;
1037 }
1038 
Commit()1039 int RelationalSyncAbleStorage::Commit()
1040 {
1041     std::unique_lock<std::shared_mutex> lock(transactionMutex_);
1042     if (transactionHandle_ == nullptr) {
1043         LOGE("relation database is null or the transaction has not been started");
1044         return -E_INVALID_DB;
1045     }
1046     int errCode = transactionHandle_->Commit();
1047     ReleaseHandle(transactionHandle_);
1048     transactionHandle_ = nullptr;
1049     LOGD("connection commit transaction!");
1050     return errCode;
1051 }
1052 
Rollback()1053 int RelationalSyncAbleStorage::Rollback()
1054 {
1055     std::unique_lock<std::shared_mutex> lock(transactionMutex_);
1056     if (transactionHandle_ == nullptr) {
1057         LOGE("Invalid handle for rollback or the transaction has not been started.");
1058         return -E_INVALID_DB;
1059     }
1060 
1061     int errCode = transactionHandle_->Rollback();
1062     ReleaseHandle(transactionHandle_);
1063     transactionHandle_ = nullptr;
1064     LOGI("connection rollback transaction!");
1065     return errCode;
1066 }
1067 
GetAllUploadCount(const QuerySyncObject & query,const std::vector<Timestamp> & timestampVec,bool isCloudForcePush,bool isCompensatedTask,int64_t & count)1068 int RelationalSyncAbleStorage::GetAllUploadCount(const QuerySyncObject &query,
1069     const std::vector<Timestamp> &timestampVec, bool isCloudForcePush, bool isCompensatedTask, int64_t &count)
1070 {
1071     int errCode = E_OK;
1072     auto *handle = GetHandleExpectTransaction(false, errCode);
1073     if (handle == nullptr) {
1074         return errCode;
1075     }
1076     QuerySyncObject queryObj = query;
1077     queryObj.SetSchema(GetSchemaInfo());
1078     errCode = handle->GetAllUploadCount(timestampVec, isCloudForcePush, isCompensatedTask, queryObj, count);
1079     if (transactionHandle_ == nullptr) {
1080         ReleaseHandle(handle);
1081     }
1082     return errCode;
1083 }
1084 
GetUploadCount(const QuerySyncObject & query,const Timestamp & timestamp,bool isCloudForcePush,bool isCompensatedTask,int64_t & count)1085 int RelationalSyncAbleStorage::GetUploadCount(const QuerySyncObject &query, const Timestamp &timestamp,
1086     bool isCloudForcePush, bool isCompensatedTask, int64_t &count)
1087 {
1088     int errCode = E_OK;
1089     auto *handle = GetHandleExpectTransaction(false, errCode);
1090     if (handle == nullptr) {
1091         return errCode;
1092     }
1093     QuerySyncObject queryObj = query;
1094     queryObj.SetSchema(GetSchemaInfo());
1095     errCode = handle->GetUploadCount(timestamp, isCloudForcePush, isCompensatedTask, queryObj, count);
1096     if (transactionHandle_ == nullptr) {
1097         ReleaseHandle(handle);
1098     }
1099     return errCode;
1100 }
1101 
GetCloudData(const TableSchema & tableSchema,const QuerySyncObject & querySyncObject,const Timestamp & beginTime,ContinueToken & continueStmtToken,CloudSyncData & cloudDataResult)1102 int RelationalSyncAbleStorage::GetCloudData(const TableSchema &tableSchema, const QuerySyncObject &querySyncObject,
1103     const Timestamp &beginTime, ContinueToken &continueStmtToken, CloudSyncData &cloudDataResult)
1104 {
1105     if (transactionHandle_ == nullptr) {
1106         LOGE("the transaction has not been started");
1107         return -E_INVALID_DB;
1108     }
1109     SyncTimeRange syncTimeRange = { .beginTime = beginTime };
1110     QuerySyncObject query = querySyncObject;
1111     query.SetSchema(GetSchemaInfo());
1112     auto token = new (std::nothrow) SQLiteSingleVerRelationalContinueToken(syncTimeRange, query);
1113     if (token == nullptr) {
1114         LOGE("[SingleVerNStore] Allocate continue token failed.");
1115         return -E_OUT_OF_MEMORY;
1116     }
1117     token->SetCloudTableSchema(tableSchema);
1118     continueStmtToken = static_cast<ContinueToken>(token);
1119     return GetCloudDataNext(continueStmtToken, cloudDataResult);
1120 }
1121 
GetCloudDataNext(ContinueToken & continueStmtToken,CloudSyncData & cloudDataResult)1122 int RelationalSyncAbleStorage::GetCloudDataNext(ContinueToken &continueStmtToken,
1123     CloudSyncData &cloudDataResult)
1124 {
1125     if (continueStmtToken == nullptr) {
1126         return -E_INVALID_ARGS;
1127     }
1128     auto token = static_cast<SQLiteSingleVerRelationalContinueToken *>(continueStmtToken);
1129     if (!token->CheckValid()) {
1130         return -E_INVALID_ARGS;
1131     }
1132     if (transactionHandle_ == nullptr) {
1133         LOGE("the transaction has not been started, release the token");
1134         ReleaseCloudDataToken(continueStmtToken);
1135         return -E_INVALID_DB;
1136     }
1137     cloudDataResult.isShared = IsSharedTable(cloudDataResult.tableName);
1138     auto config = GetCloudSyncConfig();
1139     transactionHandle_->SetUploadConfig(config.maxUploadCount, config.maxUploadSize);
1140     int errCode = transactionHandle_->GetSyncCloudData(uploadRecorder_, cloudDataResult, *token);
1141     LOGI("mode:%d upload data, ins:%zu, upd:%zu, del:%zu, lock:%zu", cloudDataResult.mode,
1142         cloudDataResult.insData.extend.size(), cloudDataResult.updData.extend.size(),
1143         cloudDataResult.delData.extend.size(), cloudDataResult.lockData.extend.size());
1144     if (errCode != -E_UNFINISHED) {
1145         delete token;
1146         token = nullptr;
1147     }
1148     continueStmtToken = static_cast<ContinueToken>(token);
1149     if (errCode != E_OK && errCode != -E_UNFINISHED) {
1150         return errCode;
1151     }
1152     int fillRefGidCode = FillReferenceData(cloudDataResult);
1153     return fillRefGidCode == E_OK ? errCode : fillRefGidCode;
1154 }
1155 
GetCloudGid(const TableSchema & tableSchema,const QuerySyncObject & querySyncObject,bool isCloudForcePush,bool isCompensatedTask,std::vector<std::string> & cloudGid)1156 int RelationalSyncAbleStorage::GetCloudGid(const TableSchema &tableSchema, const QuerySyncObject &querySyncObject,
1157     bool isCloudForcePush, bool isCompensatedTask, std::vector<std::string> &cloudGid)
1158 {
1159     int errCode = E_OK;
1160     auto *handle = GetHandle(false, errCode);
1161     if (handle == nullptr) {
1162         return errCode;
1163     }
1164     Timestamp beginTime = 0u;
1165     SyncTimeRange syncTimeRange = { .beginTime = beginTime };
1166     QuerySyncObject query = querySyncObject;
1167     query.SetSchema(GetSchemaInfo());
1168     handle->SetTableSchema(tableSchema);
1169     errCode = handle->GetSyncCloudGid(query, syncTimeRange, isCloudForcePush, isCompensatedTask, cloudGid);
1170     ReleaseHandle(handle);
1171     if (errCode != E_OK) {
1172         LOGE("[RelationalSyncAbleStorage] GetCloudGid failed %d", errCode);
1173     }
1174     return errCode;
1175 }
1176 
ReleaseCloudDataToken(ContinueToken & continueStmtToken)1177 int RelationalSyncAbleStorage::ReleaseCloudDataToken(ContinueToken &continueStmtToken)
1178 {
1179     if (continueStmtToken == nullptr) {
1180         return E_OK;
1181     }
1182     auto token = static_cast<SQLiteSingleVerRelationalContinueToken *>(continueStmtToken);
1183     if (!token->CheckValid()) {
1184         return E_OK;
1185     }
1186     int errCode = token->ReleaseCloudStatement();
1187     delete token;
1188     token = nullptr;
1189     return errCode;
1190 }
1191 
GetSchemaFromDB(RelationalSchemaObject & schema)1192 int RelationalSyncAbleStorage::GetSchemaFromDB(RelationalSchemaObject &schema)
1193 {
1194     Key schemaKey;
1195     DBCommon::StringToVector(DBConstant::RELATIONAL_SCHEMA_KEY, schemaKey);
1196     Value schemaVal;
1197     int errCode = GetMetaData(schemaKey, schemaVal);
1198     if (errCode != E_OK && errCode != -E_NOT_FOUND) {
1199         LOGE("Get relational schema from DB failed. %d", errCode);
1200         return errCode;
1201     } else if (errCode == -E_NOT_FOUND || schemaVal.empty()) {
1202         LOGW("No relational schema info was found. error %d size %zu", errCode, schemaVal.size());
1203         return -E_NOT_FOUND;
1204     }
1205     std::string schemaStr;
1206     DBCommon::VectorToString(schemaVal, schemaStr);
1207     errCode = schema.ParseFromSchemaString(schemaStr);
1208     if (errCode != E_OK) {
1209         LOGE("Parse schema string from DB failed.");
1210         return errCode;
1211     }
1212     storageEngine_->SetSchema(schema);
1213     return errCode;
1214 }
1215 
ChkSchema(const TableName & tableName)1216 int RelationalSyncAbleStorage::ChkSchema(const TableName &tableName)
1217 {
1218     std::shared_lock<std::shared_mutex> readLock(schemaMgrMutex_);
1219     RelationalSchemaObject localSchema = GetSchemaInfo();
1220     int errCode = schemaMgr_.ChkSchema(tableName, localSchema);
1221     if (errCode == -E_SCHEMA_MISMATCH) {
1222         LOGI("Get schema by tableName %s failed.", DBCommon::STR_MASK(tableName));
1223         RelationalSchemaObject newSchema;
1224         errCode = GetSchemaFromDB(newSchema);
1225         if (errCode != E_OK) {
1226             LOGE("Get schema from db when check schema. err: %d", errCode);
1227             return -E_SCHEMA_MISMATCH;
1228         }
1229         errCode = schemaMgr_.ChkSchema(tableName, newSchema);
1230     }
1231     return errCode;
1232 }
1233 
SetCloudDbSchema(const DataBaseSchema & schema)1234 int RelationalSyncAbleStorage::SetCloudDbSchema(const DataBaseSchema &schema)
1235 {
1236     std::unique_lock<std::shared_mutex> writeLock(schemaMgrMutex_);
1237     RelationalSchemaObject localSchema = GetSchemaInfo();
1238     schemaMgr_.SetCloudDbSchema(schema, localSchema);
1239     return E_OK;
1240 }
1241 
GetInfoByPrimaryKeyOrGid(const std::string & tableName,const VBucket & vBucket,DataInfoWithLog & dataInfoWithLog,VBucket & assetInfo)1242 int RelationalSyncAbleStorage::GetInfoByPrimaryKeyOrGid(const std::string &tableName, const VBucket &vBucket,
1243     DataInfoWithLog &dataInfoWithLog, VBucket &assetInfo)
1244 {
1245     if (transactionHandle_ == nullptr) {
1246         LOGE(" the transaction has not been started");
1247         return -E_INVALID_DB;
1248     }
1249 
1250     return GetInfoByPrimaryKeyOrGidInner(transactionHandle_, tableName, vBucket, dataInfoWithLog, assetInfo);
1251 }
1252 
GetInfoByPrimaryKeyOrGidInner(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,const VBucket & vBucket,DataInfoWithLog & dataInfoWithLog,VBucket & assetInfo)1253 int RelationalSyncAbleStorage::GetInfoByPrimaryKeyOrGidInner(SQLiteSingleVerRelationalStorageExecutor *handle,
1254     const std::string &tableName, const VBucket &vBucket, DataInfoWithLog &dataInfoWithLog, VBucket &assetInfo)
1255 {
1256     if (handle == nullptr) {
1257         return -E_INVALID_DB;
1258     }
1259     TableSchema tableSchema;
1260     int errCode = GetCloudTableSchema(tableName, tableSchema);
1261     if (errCode != E_OK) {
1262         LOGE("Get cloud schema failed when query log for cloud sync, %d", errCode);
1263         return errCode;
1264     }
1265     RelationalSchemaObject localSchema = GetSchemaInfo();
1266     handle->SetLocalSchema(localSchema);
1267     return handle->GetInfoByPrimaryKeyOrGid(tableSchema, vBucket, dataInfoWithLog, assetInfo);
1268 }
1269 
PutCloudSyncData(const std::string & tableName,DownloadData & downloadData)1270 int RelationalSyncAbleStorage::PutCloudSyncData(const std::string &tableName, DownloadData &downloadData)
1271 {
1272     if (transactionHandle_ == nullptr) {
1273         LOGE(" the transaction has not been started");
1274         return -E_INVALID_DB;
1275     }
1276     return PutCloudSyncDataInner(transactionHandle_, tableName, downloadData);
1277 }
1278 
PutCloudSyncDataInner(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,DownloadData & downloadData)1279 int RelationalSyncAbleStorage::PutCloudSyncDataInner(SQLiteSingleVerRelationalStorageExecutor *handle,
1280     const std::string &tableName, DownloadData &downloadData)
1281 {
1282     TableSchema tableSchema;
1283     int errCode = GetCloudTableSchema(tableName, tableSchema);
1284     if (errCode != E_OK) {
1285         LOGE("Get cloud schema failed when save cloud data, %d", errCode);
1286         return errCode;
1287     }
1288     RelationalSchemaObject localSchema = GetSchemaInfo();
1289     handle->SetLocalSchema(localSchema);
1290     TrackerTable trackerTable = storageEngine_->GetTrackerSchema().GetTrackerTable(tableName);
1291     handle->SetLogicDelete(IsCurrentLogicDelete());
1292     errCode = handle->PutCloudSyncData(tableName, tableSchema, trackerTable, downloadData);
1293     handle->SetLogicDelete(false);
1294     return errCode;
1295 }
1296 
GetCloudDbSchema(std::shared_ptr<DataBaseSchema> & cloudSchema)1297 int RelationalSyncAbleStorage::GetCloudDbSchema(std::shared_ptr<DataBaseSchema> &cloudSchema)
1298 {
1299     std::shared_lock<std::shared_mutex> readLock(schemaMgrMutex_);
1300     cloudSchema = schemaMgr_.GetCloudDbSchema();
1301     return E_OK;
1302 }
1303 
CleanCloudData(ClearMode mode,const std::vector<std::string> & tableNameList,const RelationalSchemaObject & localSchema,std::vector<Asset> & assets)1304 int RelationalSyncAbleStorage::CleanCloudData(ClearMode mode, const std::vector<std::string> &tableNameList,
1305     const RelationalSchemaObject &localSchema, std::vector<Asset> &assets)
1306 {
1307     if (transactionHandle_ == nullptr) {
1308         LOGE("the transaction has not been started");
1309         return -E_INVALID_DB;
1310     }
1311     transactionHandle_->SetLogicDelete(logicDelete_);
1312     std::vector<std::string> notifyTableList;
1313     int errCode = transactionHandle_->DoCleanInner(mode, tableNameList, localSchema, assets, notifyTableList);
1314     if (!notifyTableList.empty()) {
1315         for (auto notifyTableName : notifyTableList) {
1316             ChangedData changedData;
1317             changedData.type = ChangedDataType::DATA;
1318             changedData.tableName = notifyTableName;
1319             std::vector<DistributedDB::Type> dataVec;
1320             DistributedDB::Type type;
1321             if (mode == FLAG_ONLY) {
1322                 type = std::string(CloudDbConstant::FLAG_ONLY_MODE_NOTIFY);
1323             } else {
1324                 type = std::string(CloudDbConstant::FLAG_AND_DATA_MODE_NOTIFY);
1325             }
1326             dataVec.push_back(type);
1327             changedData.primaryData[ChangeType::OP_DELETE].push_back(dataVec);
1328             TriggerObserverAction("CLOUD", std::move(changedData), true);
1329         }
1330     }
1331     transactionHandle_->SetLogicDelete(false);
1332     return errCode;
1333 }
1334 
GetCloudTableSchema(const TableName & tableName,TableSchema & tableSchema)1335 int RelationalSyncAbleStorage::GetCloudTableSchema(const TableName &tableName, TableSchema &tableSchema)
1336 {
1337     std::shared_lock<std::shared_mutex> readLock(schemaMgrMutex_);
1338     return schemaMgr_.GetCloudTableSchema(tableName, tableSchema);
1339 }
1340 
FillCloudAssetForDownload(const std::string & tableName,VBucket & asset,bool isDownloadSuccess)1341 int RelationalSyncAbleStorage::FillCloudAssetForDownload(const std::string &tableName, VBucket &asset,
1342     bool isDownloadSuccess)
1343 {
1344     if (storageEngine_ == nullptr) {
1345         return -E_INVALID_DB;
1346     }
1347     if (transactionHandle_ == nullptr) {
1348         LOGE("the transaction has not been started when fill asset for download.");
1349         return -E_INVALID_DB;
1350     }
1351     TableSchema tableSchema;
1352     int errCode = GetCloudTableSchema(tableName, tableSchema);
1353     if (errCode != E_OK) {
1354         LOGE("Get cloud schema failed when fill cloud asset, %d", errCode);
1355         return errCode;
1356     }
1357     errCode = transactionHandle_->FillCloudAssetForDownload(tableSchema, asset, isDownloadSuccess);
1358     if (errCode != E_OK) {
1359         LOGE("fill cloud asset for download failed.%d", errCode);
1360     }
1361     return errCode;
1362 }
1363 
SetLogTriggerStatus(bool status)1364 int RelationalSyncAbleStorage::SetLogTriggerStatus(bool status)
1365 {
1366     int errCode = E_OK;
1367     auto *handle = GetHandleExpectTransaction(false, errCode);
1368     if (handle == nullptr) {
1369         return errCode;
1370     }
1371     errCode = handle->SetLogTriggerStatus(status);
1372     if (transactionHandle_ == nullptr) {
1373         ReleaseHandle(handle);
1374     }
1375     return errCode;
1376 }
1377 
SetCursorIncFlag(bool flag)1378 int RelationalSyncAbleStorage::SetCursorIncFlag(bool flag)
1379 {
1380     int errCode = E_OK;
1381     auto *handle = GetHandleExpectTransaction(false, errCode);
1382     if (handle == nullptr) {
1383         return errCode;
1384     }
1385     errCode = handle->SetCursorIncFlag(flag);
1386     if (transactionHandle_ == nullptr) {
1387         ReleaseHandle(handle);
1388     }
1389     return errCode;
1390 }
1391 
FillCloudLogAndAsset(const OpType opType,const CloudSyncData & data,bool fillAsset,bool ignoreEmptyGid)1392 int RelationalSyncAbleStorage::FillCloudLogAndAsset(const OpType opType, const CloudSyncData &data, bool fillAsset,
1393     bool ignoreEmptyGid)
1394 {
1395     if (storageEngine_ == nullptr) {
1396         return -E_INVALID_DB;
1397     }
1398     int errCode = E_OK;
1399     auto writeHandle = static_cast<SQLiteSingleVerRelationalStorageExecutor *>(
1400         storageEngine_->FindExecutor(true, OperatePerm::NORMAL_PERM, errCode));
1401     if (writeHandle == nullptr) {
1402         return errCode;
1403     }
1404     errCode = writeHandle->StartTransaction(TransactType::IMMEDIATE);
1405     if (errCode != E_OK) {
1406         ReleaseHandle(writeHandle);
1407         return errCode;
1408     }
1409     errCode = FillCloudLogAndAssetInner(writeHandle, opType, data, fillAsset, ignoreEmptyGid);
1410     if (errCode != E_OK) {
1411         LOGE("Failed to fill version or cloud asset, opType:%d ret:%d.", opType, errCode);
1412         writeHandle->Rollback();
1413         ReleaseHandle(writeHandle);
1414         return errCode;
1415     }
1416     errCode = writeHandle->Commit();
1417     ReleaseHandle(writeHandle);
1418     return errCode;
1419 }
1420 
SetSyncAbleEngine(std::shared_ptr<SyncAbleEngine> syncAbleEngine)1421 void RelationalSyncAbleStorage::SetSyncAbleEngine(std::shared_ptr<SyncAbleEngine> syncAbleEngine)
1422 {
1423     syncAbleEngine_ = syncAbleEngine;
1424 }
1425 
GetIdentify() const1426 std::string RelationalSyncAbleStorage::GetIdentify() const
1427 {
1428     if (storageEngine_ == nullptr) {
1429         LOGW("[RelationalSyncAbleStorage] engine is nullptr return default");
1430         return "";
1431     }
1432     return storageEngine_->GetIdentifier();
1433 }
1434 
EraseDataChangeCallback(uint64_t connectionId)1435 void RelationalSyncAbleStorage::EraseDataChangeCallback(uint64_t connectionId)
1436 {
1437     TaskHandle handle = ConcurrentAdapter::ScheduleTaskH([this, connectionId] () mutable {
1438         ADAPTER_AUTO_LOCK(lock, dataChangeDeviceMutex_);
1439         auto it = dataChangeCallbackMap_.find(connectionId);
1440         if (it != dataChangeCallbackMap_.end()) {
1441             dataChangeCallbackMap_.erase(it);
1442             LOGI("erase all observer for this delegate.");
1443         }
1444     }, nullptr, &dataChangeCallbackMap_);
1445     ADAPTER_WAIT(handle);
1446 }
1447 
ReleaseContinueToken(ContinueToken & continueStmtToken) const1448 void RelationalSyncAbleStorage::ReleaseContinueToken(ContinueToken &continueStmtToken) const
1449 {
1450     auto token = static_cast<SQLiteSingleVerRelationalContinueToken *>(continueStmtToken);
1451     if (token == nullptr || !(token->CheckValid())) {
1452         LOGW("[RelationalSyncAbleStorage][ReleaseContinueToken] Input is not a continue token.");
1453         return;
1454     }
1455     delete token;
1456     continueStmtToken = nullptr;
1457 }
1458 
CheckQueryValid(const QuerySyncObject & query)1459 int RelationalSyncAbleStorage::CheckQueryValid(const QuerySyncObject &query)
1460 {
1461     int errCode = E_OK;
1462     auto *handle = GetHandle(false, errCode);
1463     if (handle == nullptr) {
1464         return errCode;
1465     }
1466     errCode = handle->CheckQueryObjectLegal(query);
1467     if (errCode != E_OK) {
1468         ReleaseHandle(handle);
1469         return errCode;
1470     }
1471     QuerySyncObject queryObj = query;
1472     queryObj.SetSchema(GetSchemaInfo());
1473     int64_t count = 0;
1474     errCode = handle->GetUploadCount(UINT64_MAX, false, false, queryObj, count);
1475     ReleaseHandle(handle);
1476     if (errCode != E_OK) {
1477         LOGE("[RelationalSyncAbleStorage] CheckQueryValid failed %d", errCode);
1478         return -E_INVALID_ARGS;
1479     }
1480     return errCode;
1481 }
1482 
CreateTempSyncTrigger(const std::string & tableName)1483 int RelationalSyncAbleStorage::CreateTempSyncTrigger(const std::string &tableName)
1484 {
1485     int errCode = E_OK;
1486     auto *handle = GetHandle(true, errCode);
1487     if (handle == nullptr) {
1488         return errCode;
1489     }
1490     errCode = CreateTempSyncTriggerInner(handle, tableName, true);
1491     ReleaseHandle(handle);
1492     if (errCode != E_OK) {
1493         LOGE("[RelationalSyncAbleStorage] Create temp sync trigger failed %d", errCode);
1494     }
1495     return errCode;
1496 }
1497 
GetAndResetServerObserverData(const std::string & tableName,ChangeProperties & changeProperties)1498 int RelationalSyncAbleStorage::GetAndResetServerObserverData(const std::string &tableName,
1499     ChangeProperties &changeProperties)
1500 {
1501     int errCode = E_OK;
1502     auto *handle = GetHandle(false, errCode);
1503     if (handle == nullptr) {
1504         return errCode;
1505     }
1506     errCode = handle->GetAndResetServerObserverData(tableName, changeProperties);
1507     ReleaseHandle(handle);
1508     if (errCode != E_OK) {
1509         LOGE("[RelationalSyncAbleStorage] get server observer data failed %d", errCode);
1510     }
1511     return errCode;
1512 }
1513 
FilterChangeDataByDetailsType(ChangedData & changedData,uint32_t type)1514 void RelationalSyncAbleStorage::FilterChangeDataByDetailsType(ChangedData &changedData, uint32_t type)
1515 {
1516     if ((type & static_cast<uint32_t>(CallbackDetailsType::DEFAULT)) == 0) {
1517         changedData.field = {};
1518         for (size_t i = ChangeType::OP_INSERT; i < ChangeType::OP_BUTT; ++i) {
1519             changedData.primaryData[i].clear();
1520         }
1521     }
1522     if ((type & static_cast<uint32_t>(CallbackDetailsType::BRIEF)) == 0) {
1523         changedData.properties = {};
1524     }
1525 }
1526 
ClearAllTempSyncTrigger()1527 int RelationalSyncAbleStorage::ClearAllTempSyncTrigger()
1528 {
1529     int errCode = E_OK;
1530     auto *handle = GetHandle(true, errCode);
1531     if (handle == nullptr) {
1532         return errCode;
1533     }
1534     errCode = handle->ClearAllTempSyncTrigger();
1535     ReleaseHandle(handle);
1536     if (errCode != E_OK) {
1537         LOGE("[RelationalSyncAbleStorage] clear all temp sync trigger failed %d", errCode);
1538     }
1539     return errCode;
1540 }
1541 
FillReferenceData(CloudSyncData & syncData)1542 int RelationalSyncAbleStorage::FillReferenceData(CloudSyncData &syncData)
1543 {
1544     std::map<int64_t, Entries> referenceGid;
1545     int errCode = GetReferenceGid(syncData.tableName, syncData.insData, referenceGid);
1546     if (errCode != E_OK) {
1547         LOGE("[RelationalSyncAbleStorage] get insert reference data failed %d", errCode);
1548         return errCode;
1549     }
1550     errCode = FillReferenceDataIntoExtend(syncData.insData.rowid, referenceGid, syncData.insData.extend);
1551     if (errCode != E_OK) {
1552         return errCode;
1553     }
1554     referenceGid.clear();
1555     errCode = GetReferenceGid(syncData.tableName, syncData.updData, referenceGid);
1556     if (errCode != E_OK) {
1557         LOGE("[RelationalSyncAbleStorage] get update reference data failed %d", errCode);
1558         return errCode;
1559     }
1560     return FillReferenceDataIntoExtend(syncData.updData.rowid, referenceGid, syncData.updData.extend);
1561 }
1562 
FillReferenceDataIntoExtend(const std::vector<int64_t> & rowid,const std::map<int64_t,Entries> & referenceGid,std::vector<VBucket> & extend)1563 int RelationalSyncAbleStorage::FillReferenceDataIntoExtend(const std::vector<int64_t> &rowid,
1564     const std::map<int64_t, Entries> &referenceGid, std::vector<VBucket> &extend)
1565 {
1566     if (referenceGid.empty()) {
1567         return E_OK;
1568     }
1569     int ignoredCount = 0;
1570     for (size_t index = 0u; index < rowid.size(); index++) {
1571         if (index >= extend.size()) {
1572             LOGE("[RelationalSyncAbleStorage] index out of range when fill reference gid into extend!");
1573             return -E_UNEXPECTED_DATA;
1574         }
1575         int64_t rowId = rowid[index];
1576         if (referenceGid.find(rowId) == referenceGid.end()) {
1577             // current data miss match reference data, we ignored it
1578             ignoredCount++;
1579             continue;
1580         }
1581         extend[index].insert({ CloudDbConstant::REFERENCE_FIELD, referenceGid.at(rowId) });
1582     }
1583     if (ignoredCount != 0) {
1584         LOGD("[RelationalSyncAbleStorage] ignored %d data when fill reference data", ignoredCount);
1585     }
1586     return E_OK;
1587 }
1588 
IsSharedTable(const std::string & tableName)1589 bool RelationalSyncAbleStorage::IsSharedTable(const std::string &tableName)
1590 {
1591     std::unique_lock<std::shared_mutex> writeLock(schemaMgrMutex_);
1592     return schemaMgr_.IsSharedTable(tableName);
1593 }
1594 
GetSharedTableOriginNames()1595 std::map<std::string, std::string> RelationalSyncAbleStorage::GetSharedTableOriginNames()
1596 {
1597     std::unique_lock<std::shared_mutex> writeLock(schemaMgrMutex_);
1598     return schemaMgr_.GetSharedTableOriginNames();
1599 }
1600 
GetReferenceGid(const std::string & tableName,const CloudSyncBatch & syncBatch,std::map<int64_t,Entries> & referenceGid)1601 int RelationalSyncAbleStorage::GetReferenceGid(const std::string &tableName, const CloudSyncBatch &syncBatch,
1602     std::map<int64_t, Entries> &referenceGid)
1603 {
1604     std::map<std::string, std::vector<TableReferenceProperty>> tableReference;
1605     int errCode = GetTableReference(tableName, tableReference);
1606     if (errCode != E_OK) {
1607         return errCode;
1608     }
1609     if (tableReference.empty()) {
1610         LOGD("[RelationalSyncAbleStorage] currentTable not exist reference property");
1611         return E_OK;
1612     }
1613     auto *handle = GetHandle(false, errCode);
1614     if (handle == nullptr) {
1615         return errCode;
1616     }
1617     errCode = handle->GetReferenceGid(tableName, syncBatch, tableReference, referenceGid);
1618     ReleaseHandle(handle);
1619     return errCode;
1620 }
1621 
GetTableReference(const std::string & tableName,std::map<std::string,std::vector<TableReferenceProperty>> & reference)1622 int RelationalSyncAbleStorage::GetTableReference(const std::string &tableName,
1623     std::map<std::string, std::vector<TableReferenceProperty>> &reference)
1624 {
1625     if (storageEngine_ == nullptr) {
1626         LOGE("[RelationalSyncAbleStorage] storage is null when get reference gid");
1627         return -E_INVALID_DB;
1628     }
1629     RelationalSchemaObject schema = storageEngine_->GetSchema();
1630     auto referenceProperty = schema.GetReferenceProperty();
1631     if (referenceProperty.empty()) {
1632         return E_OK;
1633     }
1634     auto [sourceTableName, errCode] = GetSourceTableName(tableName);
1635     if (errCode != E_OK) {
1636         return errCode;
1637     }
1638     for (const auto &property : referenceProperty) {
1639         if (DBCommon::CaseInsensitiveCompare(property.sourceTableName, sourceTableName)) {
1640             if (!IsSharedTable(tableName)) {
1641                 reference[property.targetTableName].push_back(property);
1642                 continue;
1643             }
1644             TableReferenceProperty tableReference;
1645             tableReference.sourceTableName = tableName;
1646             tableReference.columns = property.columns;
1647             tableReference.columns[CloudDbConstant::CLOUD_OWNER] = CloudDbConstant::CLOUD_OWNER;
1648             auto [sharedTargetTable, ret] = GetSharedTargetTableName(property.targetTableName);
1649             if (ret != E_OK) {
1650                 return ret;
1651             }
1652             tableReference.targetTableName = sharedTargetTable;
1653             reference[tableReference.targetTableName].push_back(tableReference);
1654         }
1655     }
1656     return E_OK;
1657 }
1658 
GetSourceTableName(const std::string & tableName)1659 std::pair<std::string, int> RelationalSyncAbleStorage::GetSourceTableName(const std::string &tableName)
1660 {
1661     std::pair<std::string, int> res = { "", E_OK };
1662     std::shared_ptr<DataBaseSchema> cloudSchema;
1663     (void) GetCloudDbSchema(cloudSchema);
1664     if (cloudSchema == nullptr) {
1665         LOGE("[RelationalSyncAbleStorage] cloud schema is null when get source table");
1666         return { "", -E_INTERNAL_ERROR };
1667     }
1668     for (const auto &table : cloudSchema->tables) {
1669         if (CloudStorageUtils::IsSharedTable(table)) {
1670             continue;
1671         }
1672         if (DBCommon::CaseInsensitiveCompare(table.name, tableName) ||
1673             DBCommon::CaseInsensitiveCompare(table.sharedTableName, tableName)) {
1674             res.first = table.name;
1675             break;
1676         }
1677     }
1678     if (res.first.empty()) {
1679         LOGE("[RelationalSyncAbleStorage] not found table in cloud schema");
1680         res.second = -E_SCHEMA_MISMATCH;
1681     }
1682     return res;
1683 }
1684 
GetSharedTargetTableName(const std::string & tableName)1685 std::pair<std::string, int> RelationalSyncAbleStorage::GetSharedTargetTableName(const std::string &tableName)
1686 {
1687     std::pair<std::string, int> res = { "", E_OK };
1688     std::shared_ptr<DataBaseSchema> cloudSchema;
1689     (void) GetCloudDbSchema(cloudSchema);
1690     if (cloudSchema == nullptr) {
1691         LOGE("[RelationalSyncAbleStorage] cloud schema is null when get shared target table");
1692         return { "", -E_INTERNAL_ERROR };
1693     }
1694     for (const auto &table : cloudSchema->tables) {
1695         if (CloudStorageUtils::IsSharedTable(table)) {
1696             continue;
1697         }
1698         if (DBCommon::CaseInsensitiveCompare(table.name, tableName)) {
1699             res.first = table.sharedTableName;
1700             break;
1701         }
1702     }
1703     if (res.first.empty()) {
1704         LOGE("[RelationalSyncAbleStorage] not found table in cloud schema");
1705         res.second = -E_SCHEMA_MISMATCH;
1706     }
1707     return res;
1708 }
1709 
SetLogicDelete(bool logicDelete)1710 void RelationalSyncAbleStorage::SetLogicDelete(bool logicDelete)
1711 {
1712     logicDelete_ = logicDelete;
1713     LOGI("[RelationalSyncAbleStorage] set logic delete %d", static_cast<int>(logicDelete));
1714 }
1715 
SetCloudTaskConfig(const CloudTaskConfig & config)1716 void RelationalSyncAbleStorage::SetCloudTaskConfig(const CloudTaskConfig &config)
1717 {
1718     allowLogicDelete_ = config.allowLogicDelete;
1719     LOGD("[RelationalSyncAbleStorage] allow logic delete %d", static_cast<int>(config.allowLogicDelete));
1720 }
1721 
IsCurrentLogicDelete() const1722 bool RelationalSyncAbleStorage::IsCurrentLogicDelete() const
1723 {
1724     return allowLogicDelete_ && logicDelete_;
1725 }
1726 
GetAssetsByGidOrHashKey(const TableSchema & tableSchema,const std::string & gid,const Bytes & hashKey,VBucket & assets)1727 std::pair<int, uint32_t> RelationalSyncAbleStorage::GetAssetsByGidOrHashKey(const TableSchema &tableSchema,
1728     const std::string &gid, const Bytes &hashKey, VBucket &assets)
1729 {
1730     if (gid.empty() && hashKey.empty()) {
1731         LOGE("both gid and hashKey are empty.");
1732         return { -E_INVALID_ARGS, static_cast<uint32_t>(LockStatus::UNLOCK) };
1733     }
1734     if (transactionHandle_ == nullptr) {
1735         LOGE("the transaction has not been started");
1736         return { -E_INVALID_DB, static_cast<uint32_t>(LockStatus::UNLOCK) };
1737     }
1738     auto [errCode, status] = transactionHandle_->GetAssetsByGidOrHashKey(tableSchema, gid, hashKey, assets);
1739     if (errCode != E_OK && errCode != -E_NOT_FOUND && errCode != -E_CLOUD_GID_MISMATCH) {
1740         LOGE("get assets by gid or hashKey failed. %d", errCode);
1741     }
1742     return { errCode, status };
1743 }
1744 
SetIAssetLoader(const std::shared_ptr<IAssetLoader> & loader)1745 int RelationalSyncAbleStorage::SetIAssetLoader(const std::shared_ptr<IAssetLoader> &loader)
1746 {
1747     int errCode = E_OK;
1748     auto *wHandle = GetHandle(true, errCode);
1749     if (wHandle == nullptr) {
1750         return errCode;
1751     }
1752     wHandle->SetIAssetLoader(loader);
1753     ReleaseHandle(wHandle);
1754     return errCode;
1755 }
1756 
UpsertData(RecordStatus status,const std::string & tableName,const std::vector<VBucket> & records)1757 int RelationalSyncAbleStorage::UpsertData(RecordStatus status, const std::string &tableName,
1758     const std::vector<VBucket> &records)
1759 {
1760     int errCode = E_OK;
1761     auto *handle = GetHandle(true, errCode);
1762     if (errCode != E_OK) {
1763         return errCode;
1764     }
1765     handle->SetPutDataMode(SQLiteSingleVerRelationalStorageExecutor::PutDataMode::USER);
1766     handle->SetMarkFlagOption(SQLiteSingleVerRelationalStorageExecutor::MarkFlagOption::SET_WAIT_COMPENSATED_SYNC);
1767     errCode = UpsertDataInner(handle, tableName, records);
1768     handle->SetPutDataMode(SQLiteSingleVerRelationalStorageExecutor::PutDataMode::SYNC);
1769     handle->SetMarkFlagOption(SQLiteSingleVerRelationalStorageExecutor::MarkFlagOption::DEFAULT);
1770     ReleaseHandle(handle);
1771     return errCode;
1772 }
1773 
UpsertDataInner(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,const std::vector<VBucket> & records)1774 int RelationalSyncAbleStorage::UpsertDataInner(SQLiteSingleVerRelationalStorageExecutor *handle,
1775     const std::string &tableName, const std::vector<VBucket> &records)
1776 {
1777     int errCode = E_OK;
1778     errCode = handle->StartTransaction(TransactType::IMMEDIATE);
1779     if (errCode != E_OK) {
1780         LOGE("[RDBStorageEngine] start transaction failed %d when upsert data", errCode);
1781         return errCode;
1782     }
1783     errCode = CreateTempSyncTriggerInner(handle, tableName);
1784     if (errCode == E_OK) {
1785         errCode = UpsertDataInTransaction(handle, tableName, records);
1786         (void) handle->ClearAllTempSyncTrigger();
1787     }
1788     if (errCode == E_OK) {
1789         errCode = handle->Commit();
1790         if (errCode != E_OK) {
1791             LOGE("[RDBStorageEngine] commit failed %d when upsert data", errCode);
1792         }
1793     } else {
1794         int ret = handle->Rollback();
1795         if (ret != E_OK) {
1796             LOGW("[RDBStorageEngine] rollback failed %d when upsert data", ret);
1797         }
1798     }
1799     return errCode;
1800 }
1801 
UpsertDataInTransaction(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,const std::vector<VBucket> & records)1802 int RelationalSyncAbleStorage::UpsertDataInTransaction(SQLiteSingleVerRelationalStorageExecutor *handle,
1803     const std::string &tableName, const std::vector<VBucket> &records)
1804 {
1805     TableSchema tableSchema;
1806     int errCode = GetCloudTableSchema(tableName, tableSchema);
1807     if (errCode != E_OK) {
1808         LOGE("Get cloud schema failed when save cloud data, %d", errCode);
1809         return errCode;
1810     }
1811     TableInfo localTable = GetSchemaInfo().GetTable(tableName); // for upsert, the table must exist in local
1812     std::map<std::string, Field> pkMap = CloudStorageUtils::GetCloudPrimaryKeyFieldMap(tableSchema, true);
1813     std::set<std::vector<uint8_t>> primaryKeys;
1814     DownloadData downloadData;
1815     for (const auto &record : records) {
1816         DataInfoWithLog dataInfoWithLog;
1817         VBucket assetInfo;
1818         auto [errorCode, hashValue] = CloudStorageUtils::GetHashValueWithPrimaryKeyMap(record,
1819             tableSchema, localTable, pkMap, false);
1820         if (errorCode != E_OK) {
1821             return errorCode;
1822         }
1823         errCode = GetInfoByPrimaryKeyOrGidInner(handle, tableName, record, dataInfoWithLog, assetInfo);
1824         if (errCode != E_OK && errCode != -E_NOT_FOUND) {
1825             return errCode;
1826         }
1827         VBucket recordCopy = record;
1828         if ((errCode == -E_NOT_FOUND ||
1829             (dataInfoWithLog.logInfo.flag & static_cast<uint32_t>(LogInfoFlag::FLAG_DELETE)) != 0) &&
1830             primaryKeys.find(hashValue) == primaryKeys.end()) {
1831             downloadData.opType.push_back(OpType::INSERT);
1832             auto currentTime = TimeHelper::GetSysCurrentTime();
1833             recordCopy[CloudDbConstant::MODIFY_FIELD] = static_cast<int64_t>(currentTime);
1834             recordCopy[CloudDbConstant::CREATE_FIELD] = static_cast<int64_t>(currentTime);
1835             primaryKeys.insert(hashValue);
1836         } else {
1837             downloadData.opType.push_back(OpType::UPDATE);
1838             recordCopy[CloudDbConstant::GID_FIELD] = dataInfoWithLog.logInfo.cloudGid;
1839             recordCopy[CloudDbConstant::MODIFY_FIELD] = static_cast<int64_t>(dataInfoWithLog.logInfo.timestamp);
1840             recordCopy[CloudDbConstant::CREATE_FIELD] = static_cast<int64_t>(dataInfoWithLog.logInfo.wTimestamp);
1841             recordCopy[CloudDbConstant::SHARING_RESOURCE_FIELD] = dataInfoWithLog.logInfo.sharingResource;
1842             recordCopy[CloudDbConstant::VERSION_FIELD] = dataInfoWithLog.logInfo.version;
1843         }
1844         downloadData.existDataKey.push_back(dataInfoWithLog.logInfo.dataKey);
1845         downloadData.data.push_back(std::move(recordCopy));
1846     }
1847     return PutCloudSyncDataInner(handle, tableName, downloadData);
1848 }
1849 
UpdateRecordFlag(const std::string & tableName,bool recordConflict,const LogInfo & logInfo)1850 int RelationalSyncAbleStorage::UpdateRecordFlag(const std::string &tableName, bool recordConflict,
1851     const LogInfo &logInfo)
1852 {
1853     if (transactionHandle_ == nullptr) {
1854         LOGE("[RelationalSyncAbleStorage] the transaction has not been started");
1855         return -E_INVALID_DB;
1856     }
1857     std::string sql = CloudStorageUtils::GetUpdateRecordFlagSql(tableName, recordConflict, logInfo);
1858     return transactionHandle_->UpdateRecordFlag(tableName, sql, logInfo);
1859 }
1860 
FillCloudLogAndAssetInner(SQLiteSingleVerRelationalStorageExecutor * handle,OpType opType,const CloudSyncData & data,bool fillAsset,bool ignoreEmptyGid)1861 int RelationalSyncAbleStorage::FillCloudLogAndAssetInner(SQLiteSingleVerRelationalStorageExecutor *handle,
1862     OpType opType, const CloudSyncData &data, bool fillAsset, bool ignoreEmptyGid)
1863 {
1864     TableSchema tableSchema;
1865     int errCode = GetCloudTableSchema(data.tableName, tableSchema);
1866     if (errCode != E_OK) {
1867         LOGE("get table schema failed when fill log and asset. %d", errCode);
1868         return errCode;
1869     }
1870     errCode = handle->FillHandleWithOpType(opType, data, fillAsset, ignoreEmptyGid, tableSchema);
1871     if (errCode != E_OK) {
1872         return errCode;
1873     }
1874     if (opType == OpType::INSERT) {
1875         errCode = UpdateRecordFlagAfterUpload(handle, data.tableName, data.insData, CloudWaterType::INSERT);
1876     } else if (opType == OpType::UPDATE) {
1877         errCode = UpdateRecordFlagAfterUpload(handle, data.tableName, data.updData, CloudWaterType::UPDATE);
1878     } else if (opType == OpType::DELETE) {
1879         errCode = UpdateRecordFlagAfterUpload(handle, data.tableName, data.delData, CloudWaterType::DELETE);
1880     } else if (opType == OpType::LOCKED_NOT_HANDLE) {
1881         errCode = UpdateRecordFlagAfterUpload(handle, data.tableName, data.lockData, CloudWaterType::BUTT, true);
1882     }
1883     return errCode;
1884 }
1885 
UpdateRecordFlagAfterUpload(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,const CloudSyncBatch & updateData,const CloudWaterType & type,bool isLock)1886 int RelationalSyncAbleStorage::UpdateRecordFlagAfterUpload(SQLiteSingleVerRelationalStorageExecutor *handle,
1887     const std::string &tableName, const CloudSyncBatch &updateData, const CloudWaterType &type, bool isLock)
1888 {
1889     if (updateData.timestamp.size() != updateData.extend.size()) {
1890         LOGE("the num of extend:%zu and timestamp:%zu is not equal.",
1891             updateData.extend.size(), updateData.timestamp.size());
1892         return -E_INVALID_ARGS;
1893     }
1894     for (size_t i = 0; i < updateData.extend.size(); ++i) {
1895         const auto &record = updateData.extend[i];
1896         if (DBCommon::IsRecordError(record) || DBCommon::IsRecordAssetsMissing(record) ||
1897             DBCommon::IsRecordVersionConflict(record) || isLock) {
1898             if (DBCommon::IsRecordAssetsMissing(record)) {
1899                 LOGI("[RDBStorage][UpdateRecordFlagAfterUpload] Record assets missing, skip update.");
1900             }
1901             int errCode = handle->UpdateRecordStatus(tableName, CloudDbConstant::TO_LOCAL_CHANGE,
1902                 updateData.hashKey[i]);
1903             if (errCode != E_OK) {
1904                 LOGE("[RDBStorage] Update record status failed in index %zu", i);
1905                 return errCode;
1906             }
1907             continue;
1908         }
1909         const auto &rowId = updateData.rowid[i];
1910         std::string cloudGid;
1911         (void)CloudStorageUtils::GetValueFromVBucket(CloudDbConstant::GID_FIELD, record, cloudGid);
1912         LogInfo logInfo;
1913         logInfo.cloudGid = cloudGid;
1914         logInfo.timestamp = updateData.timestamp[i];
1915         logInfo.dataKey = rowId;
1916         logInfo.hashKey = updateData.hashKey[i];
1917         std::string sql = CloudStorageUtils::GetUpdateRecordFlagSqlUpload(tableName, DBCommon::IsRecordIgnored(record),
1918             logInfo, record, type);
1919         int errCode = handle->UpdateRecordFlag(tableName, sql, logInfo);
1920         if (errCode != E_OK) {
1921             LOGE("[RDBStorage] Update record flag failed in index %zu", i);
1922             return errCode;
1923         }
1924         handle->MarkFlagAsUploadFinished(tableName, updateData.hashKey[i], updateData.timestamp[i]);
1925         uploadRecorder_.RecordUploadRecord(tableName, logInfo.hashKey, type, updateData.timestamp[i]);
1926     }
1927     return E_OK;
1928 }
1929 
GetCompensatedSyncQuery(std::vector<QuerySyncObject> & syncQuery,std::vector<std::string> & users)1930 int RelationalSyncAbleStorage::GetCompensatedSyncQuery(std::vector<QuerySyncObject> &syncQuery,
1931     std::vector<std::string> &users)
1932 {
1933     std::vector<TableSchema> tables;
1934     int errCode = GetCloudTableWithoutShared(tables);
1935     if (errCode != E_OK) {
1936         return errCode;
1937     }
1938     if (tables.empty()) {
1939         LOGD("[RDBStorage] Table is empty, no need to compensated sync");
1940         return E_OK;
1941     }
1942     auto *handle = GetHandle(true, errCode);
1943     if (errCode != E_OK) {
1944         return errCode;
1945     }
1946     errCode = GetCompensatedSyncQueryInner(handle, tables, syncQuery);
1947     ReleaseHandle(handle);
1948     return errCode;
1949 }
1950 
ClearUnLockingNoNeedCompensated()1951 int RelationalSyncAbleStorage::ClearUnLockingNoNeedCompensated()
1952 {
1953     std::vector<TableSchema> tables;
1954     int errCode = GetCloudTableWithoutShared(tables);
1955     if (errCode != E_OK) {
1956         return errCode;
1957     }
1958     if (tables.empty()) {
1959         LOGI("[RDBStorage] Table is empty, no need to clear unlocking status");
1960         return E_OK;
1961     }
1962     auto *handle = GetHandle(true, errCode);
1963     if (errCode != E_OK) {
1964         return errCode;
1965     }
1966     errCode = handle->StartTransaction(TransactType::IMMEDIATE);
1967     if (errCode != E_OK) {
1968         ReleaseHandle(handle);
1969         return errCode;
1970     }
1971     for (const auto &table : tables) {
1972         errCode = handle->ClearUnLockingStatus(table.name);
1973         if (errCode != E_OK) {
1974             LOGW("[ClearUnLockingNoNeedCompensated] clear unlocking status failed, continue! errCode=%d", errCode);
1975         }
1976     }
1977     errCode = handle->Commit();
1978     if (errCode != E_OK) {
1979         LOGE("[ClearUnLockingNoNeedCompensated] commit failed %d when clear unlocking status", errCode);
1980     }
1981     ReleaseHandle(handle);
1982     return errCode;
1983 }
1984 
GetCloudTableWithoutShared(std::vector<TableSchema> & tables)1985 int RelationalSyncAbleStorage::GetCloudTableWithoutShared(std::vector<TableSchema> &tables)
1986 {
1987     const auto tableInfos = GetSchemaInfo().GetTables();
1988     for (const auto &[tableName, info] : tableInfos) {
1989         if (info.GetSharedTableMark()) {
1990             continue;
1991         }
1992         TableSchema schema;
1993         int errCode = GetCloudTableSchema(tableName, schema);
1994         if (errCode == -E_NOT_FOUND) {
1995             continue;
1996         }
1997         if (errCode != E_OK) {
1998             LOGW("[RDBStorage] Get cloud table failed %d", errCode);
1999             return errCode;
2000         }
2001         tables.push_back(schema);
2002     }
2003     return E_OK;
2004 }
2005 
GetCompensatedSyncQueryInner(SQLiteSingleVerRelationalStorageExecutor * handle,const std::vector<TableSchema> & tables,std::vector<QuerySyncObject> & syncQuery)2006 int RelationalSyncAbleStorage::GetCompensatedSyncQueryInner(SQLiteSingleVerRelationalStorageExecutor *handle,
2007     const std::vector<TableSchema> &tables, std::vector<QuerySyncObject> &syncQuery)
2008 {
2009     int errCode = E_OK;
2010     errCode = handle->StartTransaction(TransactType::IMMEDIATE);
2011     if (errCode != E_OK) {
2012         return errCode;
2013     }
2014     for (const auto &table : tables) {
2015         if (!CheckTableSupportCompensatedSync(table)) {
2016             continue;
2017         }
2018 
2019         std::vector<VBucket> syncDataPk;
2020         errCode = handle->GetWaitCompensatedSyncDataPk(table, syncDataPk);
2021         if (errCode != E_OK) {
2022             LOGW("[RDBStorageEngine] Get wait compensated sync date failed, continue! errCode=%d", errCode);
2023             errCode = E_OK;
2024             continue;
2025         }
2026         if (syncDataPk.empty()) {
2027             // no data need to compensated sync
2028             continue;
2029         }
2030         QuerySyncObject syncObject;
2031         errCode = CloudStorageUtils::GetSyncQueryByPk(table.name, syncDataPk, false, syncObject);
2032         if (errCode != E_OK) {
2033             LOGW("[RDBStorageEngine] Get compensated sync query happen error, ignore it! errCode = %d", errCode);
2034             errCode = E_OK;
2035             continue;
2036         }
2037         syncQuery.push_back(syncObject);
2038     }
2039     if (errCode == E_OK) {
2040         errCode = handle->Commit();
2041         if (errCode != E_OK) {
2042             LOGE("[RDBStorageEngine] commit failed %d when get compensated sync query", errCode);
2043         }
2044     } else {
2045         int ret = handle->Rollback();
2046         if (ret != E_OK) {
2047             LOGW("[RDBStorageEngine] rollback failed %d when get compensated sync query", ret);
2048         }
2049     }
2050     return errCode;
2051 }
2052 
CreateTempSyncTriggerInner(SQLiteSingleVerRelationalStorageExecutor * handle,const std::string & tableName,bool flag)2053 int RelationalSyncAbleStorage::CreateTempSyncTriggerInner(SQLiteSingleVerRelationalStorageExecutor *handle,
2054     const std::string &tableName, bool flag)
2055 {
2056     TrackerTable trackerTable = storageEngine_->GetTrackerSchema().GetTrackerTable(tableName);
2057     if (trackerTable.IsEmpty()) {
2058         trackerTable.SetTableName(tableName);
2059     }
2060     return handle->CreateTempSyncTrigger(trackerTable, flag);
2061 }
2062 
CheckTableSupportCompensatedSync(const TableSchema & table)2063 bool RelationalSyncAbleStorage::CheckTableSupportCompensatedSync(const TableSchema &table)
2064 {
2065     auto it = std::find_if(table.fields.begin(), table.fields.end(), [](const auto &field) {
2066         return field.primary && (field.type == TYPE_INDEX<Asset> || field.type == TYPE_INDEX<Assets> ||
2067             field.type == TYPE_INDEX<Bytes>);
2068     });
2069     if (it != table.fields.end()) {
2070         LOGI("[RDBStorageEngine] Table contain not support pk field type, ignored");
2071         return false;
2072     }
2073     // check whether reference exist
2074     std::map<std::string, std::vector<TableReferenceProperty>> tableReference;
2075     int errCode = RelationalSyncAbleStorage::GetTableReference(table.name, tableReference);
2076     if (errCode != E_OK) {
2077         LOGW("[RDBStorageEngine] Get table reference failed! errCode = %d", errCode);
2078         return false;
2079     }
2080     if (!tableReference.empty()) {
2081         LOGI("[RDBStorageEngine] current table exist reference property");
2082         return false;
2083     }
2084     return true;
2085 }
2086 
MarkFlagAsConsistent(const std::string & tableName,const DownloadData & downloadData,const std::set<std::string> & gidFilters)2087 int RelationalSyncAbleStorage::MarkFlagAsConsistent(const std::string &tableName, const DownloadData &downloadData,
2088     const std::set<std::string> &gidFilters)
2089 {
2090     if (transactionHandle_ == nullptr) {
2091         LOGE("the transaction has not been started");
2092         return -E_INVALID_DB;
2093     }
2094     int errCode = transactionHandle_->MarkFlagAsConsistent(tableName, downloadData, gidFilters);
2095     if (errCode != E_OK) {
2096         LOGE("[RelationalSyncAbleStorage] mark flag as consistent failed.%d", errCode);
2097     }
2098     return errCode;
2099 }
2100 
GetCloudSyncConfig() const2101 CloudSyncConfig RelationalSyncAbleStorage::GetCloudSyncConfig() const
2102 {
2103     std::lock_guard<std::mutex> autoLock(configMutex_);
2104     return cloudSyncConfig_;
2105 }
2106 
SetCloudSyncConfig(const CloudSyncConfig & config)2107 void RelationalSyncAbleStorage::SetCloudSyncConfig(const CloudSyncConfig &config)
2108 {
2109     std::lock_guard<std::mutex> autoLock(configMutex_);
2110     cloudSyncConfig_ = config;
2111 }
2112 
IsTableExistReference(const std::string & table)2113 bool RelationalSyncAbleStorage::IsTableExistReference(const std::string &table)
2114 {
2115     // check whether reference exist
2116     std::map<std::string, std::vector<TableReferenceProperty>> tableReference;
2117     int errCode = RelationalSyncAbleStorage::GetTableReference(table, tableReference);
2118     if (errCode != E_OK) {
2119         LOGW("[RDBStorageEngine] Get table reference failed! errCode = %d", errCode);
2120         return false;
2121     }
2122     return !tableReference.empty();
2123 }
2124 
IsTableExistReferenceOrReferenceBy(const std::string & table)2125 bool RelationalSyncAbleStorage::IsTableExistReferenceOrReferenceBy(const std::string &table)
2126 {
2127     // check whether reference or reference by exist
2128     if (storageEngine_ == nullptr) {
2129         LOGE("[IsTableExistReferenceOrReferenceBy] storage is null when get reference gid");
2130         return false;
2131     }
2132     RelationalSchemaObject schema = storageEngine_->GetSchema();
2133     auto referenceProperty = schema.GetReferenceProperty();
2134     if (referenceProperty.empty()) {
2135         return false;
2136     }
2137     auto [sourceTableName, errCode] = GetSourceTableName(table);
2138     if (errCode != E_OK) {
2139         return false;
2140     }
2141     for (const auto &property : referenceProperty) {
2142         if (DBCommon::CaseInsensitiveCompare(property.sourceTableName, sourceTableName) ||
2143             DBCommon::CaseInsensitiveCompare(property.targetTableName, sourceTableName)) {
2144             return true;
2145         }
2146     }
2147     return false;
2148 }
2149 
ReleaseUploadRecord(const std::string & tableName,const CloudWaterType & type,Timestamp localMark)2150 void RelationalSyncAbleStorage::ReleaseUploadRecord(const std::string &tableName, const CloudWaterType &type,
2151     Timestamp localMark)
2152 {
2153     uploadRecorder_.ReleaseUploadRecord(tableName, type, localMark);
2154 }
2155 }
2156 #endif