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 "db_common.h"
17 #include "generic_single_ver_kv_entry.h"
18 #include "platform_specific.h"
19 #include "relational_remote_query_continue_token.h"
20 #include "runtime_context.h"
21 #include "virtual_relational_ver_sync_db_interface.h"
22 #include "virtual_single_ver_sync_db_Interface.h"
23
24 namespace DistributedDB {
25 namespace {
GetEntriesFromItems(std::vector<SingleVerKvEntry * > & entries,const std::vector<DataItem> & dataItems)26 int GetEntriesFromItems(std::vector<SingleVerKvEntry *> &entries, const std::vector<DataItem> &dataItems)
27 {
28 int errCode = E_OK;
29 for (const auto &item : dataItems) {
30 auto entry = new (std::nothrow) GenericSingleVerKvEntry();
31 if (entry == nullptr) {
32 LOGE("Create entry failed.");
33 errCode = -E_OUT_OF_MEMORY;
34 break;
35 }
36 DataItem storageItem;
37 storageItem.key = item.key;
38 storageItem.value = item.value;
39 storageItem.flag = item.flag;
40 storageItem.timestamp = item.timestamp;
41 storageItem.writeTimestamp = item.writeTimestamp;
42 storageItem.hashKey = item.hashKey;
43 entry->SetEntryData(std::move(storageItem));
44 entries.push_back(entry);
45 }
46 if (errCode != E_OK) {
47 LOGD("[GetEntriesFromItems] failed:%d", errCode);
48 for (auto &kvEntry : entries) {
49 delete kvEntry;
50 kvEntry = nullptr;
51 }
52 entries.clear();
53 }
54 LOGD("[GetEntriesFromItems] size:%zu", dataItems.size());
55 return errCode;
56 }
57
GetStr(const std::vector<uint8_t> & vec)58 std::string GetStr(const std::vector<uint8_t> &vec)
59 {
60 std::string str;
61 DBCommon::VectorToString(vec, str);
62 return str;
63 }
64 }
65
VirtualRelationalVerSyncDBInterface()66 VirtualRelationalVerSyncDBInterface::VirtualRelationalVerSyncDBInterface()
67 {
68 (void)OS::GetCurrentSysTimeInMicrosecond(dbCreateTime_);
69 LOGD("virtual device init db createTime");
70 }
71
PutSyncDataWithQuery(const QueryObject & query,const std::vector<SingleVerKvEntry * > & entries,const std::string & deviceName)72 int VirtualRelationalVerSyncDBInterface::PutSyncDataWithQuery(const QueryObject &query,
73 const std::vector<SingleVerKvEntry *> &entries, const std::string &deviceName)
74 {
75 LOGD("[PutSyncData] size %zu", entries.size());
76 std::vector<DataItem> dataItems;
77 for (auto itemEntry : entries) {
78 auto *entry = static_cast<GenericSingleVerKvEntry *>(itemEntry);
79 if (entry != nullptr) {
80 DataItem item;
81 item.origDev = entry->GetOrigDevice();
82 item.flag = entry->GetFlag();
83 item.timestamp = entry->GetTimestamp();
84 item.writeTimestamp = entry->GetWriteTimestamp();
85 entry->GetKey(item.key);
86 entry->GetValue(item.value);
87 entry->GetHashKey(item.hashKey);
88 dataItems.push_back(item);
89 }
90 }
91 OptTableDataWithLog optTableDataWithLog;
92 optTableDataWithLog.tableName = query.GetTableName();
93 int errCode = DataTransformer::TransformDataItem(dataItems, localFieldInfo_, optTableDataWithLog);
94 if (errCode != E_OK) {
95 return errCode;
96 }
97 for (const auto &optRowDataWithLog : optTableDataWithLog.dataList) {
98 VirtualRowData virtualRowData;
99 virtualRowData.logInfo = optRowDataWithLog.logInfo;
100 size_t index = 0;
101 for (const auto &optItem : optRowDataWithLog.optionalData) {
102 if (index >= localFieldInfo_.size()) {
103 break;
104 }
105 DataValue dataValue = std::move(optItem);
106 virtualRowData.objectData.PutDataValue(localFieldInfo_[index].GetFieldName(), dataValue);
107 index++;
108 }
109 syncData_[query.GetTableName()][GetStr(virtualRowData.logInfo.hashKey)] = virtualRowData;
110 }
111 LOGD("tableName %s", optTableDataWithLog.tableName.c_str());
112 return errCode;
113 }
114
PutLocalData(const std::vector<VirtualRowData> & dataList,const std::string & tableName)115 int VirtualRelationalVerSyncDBInterface::PutLocalData(const std::vector<VirtualRowData> &dataList,
116 const std::string &tableName)
117 {
118 for (const auto &item : dataList) {
119 localData_[tableName][GetStr(item.logInfo.hashKey)] = item;
120 }
121 return E_OK;
122 }
123
GetSyncData(QueryObject & query,const SyncTimeRange & timeRange,const DataSizeSpecInfo & dataSizeInfo,ContinueToken & continueStmtToken,std::vector<SingleVerKvEntry * > & entries) const124 int VirtualRelationalVerSyncDBInterface::GetSyncData(QueryObject &query,
125 const SyncTimeRange &timeRange, const DataSizeSpecInfo &dataSizeInfo,
126 ContinueToken &continueStmtToken, std::vector<SingleVerKvEntry *> &entries) const
127 {
128 if (localData_.find(query.GetTableName()) == localData_.end()) {
129 LOGD("[GetSyncData] No Data Return");
130 return E_OK;
131 }
132 std::vector<DataItem> dataItemList;
133 TableDataWithLog tableDataWithLog = {query.GetTableName(), {}};
134 for (const auto &[hashKey, virtualData] : localData_[query.GetTableName()]) {
135 if (virtualData.logInfo.timestamp < timeRange.beginTime ||
136 virtualData.logInfo.timestamp >= timeRange.endTime) {
137 LOGD("ignore hashkey %s", DBCommon::TransferStringToHex(hashKey).c_str());
138 continue;
139 }
140 RowDataWithLog rowData;
141 for (const auto &field : localFieldInfo_) {
142 DataValue dataValue;
143 (void)virtualData.objectData.GetDataValue(field.GetFieldName(), dataValue);
144 rowData.rowData.push_back(std::move(dataValue));
145 }
146 rowData.logInfo = virtualData.logInfo;
147 tableDataWithLog.dataList.push_back(rowData);
148 }
149
150 int errCode = DataTransformer::TransformTableData(tableDataWithLog, localFieldInfo_, dataItemList);
151 if (errCode != E_OK) {
152 return errCode;
153 }
154 continueStmtToken = nullptr;
155 return GetEntriesFromItems(entries, dataItemList);
156 }
157
GetSchemaInfo() const158 RelationalSchemaObject VirtualRelationalVerSyncDBInterface::GetSchemaInfo() const
159 {
160 return schemaObj_;
161 }
162
163
SetSchemaInfo(const RelationalSchemaObject & schema)164 void VirtualRelationalVerSyncDBInterface::SetSchemaInfo(const RelationalSchemaObject &schema)
165 {
166 schemaObj_ = schema;
167 }
168
GetDatabaseCreateTimestamp(Timestamp & outTime) const169 int VirtualRelationalVerSyncDBInterface::GetDatabaseCreateTimestamp(Timestamp &outTime) const
170 {
171 outTime = dbCreateTime_;
172 return E_OK;
173 }
174
GetTablesQuery()175 std::vector<QuerySyncObject> VirtualRelationalVerSyncDBInterface::GetTablesQuery()
176 {
177 return {};
178 }
179
LocalDataChanged(int notifyEvent,std::vector<QuerySyncObject> & queryObj)180 int VirtualRelationalVerSyncDBInterface::LocalDataChanged(int notifyEvent, std::vector<QuerySyncObject> &queryObj)
181 {
182 return E_OK;
183 }
184
GetInterfaceType() const185 int VirtualRelationalVerSyncDBInterface::GetInterfaceType() const
186 {
187 return SYNC_RELATION;
188 }
189
IncRefCount()190 void VirtualRelationalVerSyncDBInterface::IncRefCount()
191 {
192 }
193
DecRefCount()194 void VirtualRelationalVerSyncDBInterface::DecRefCount()
195 {
196 }
197
GetIdentifier() const198 std::vector<uint8_t> VirtualRelationalVerSyncDBInterface::GetIdentifier() const
199 {
200 return {};
201 }
202
GetMaxTimestamp(Timestamp & stamp) const203 void VirtualRelationalVerSyncDBInterface::GetMaxTimestamp(Timestamp &stamp) const
204 {
205 for (const auto &item : syncData_) {
206 for (const auto &entry : item.second) {
207 if (stamp < entry.second.logInfo.timestamp) {
208 stamp = entry.second.logInfo.timestamp;
209 }
210 }
211 }
212 LOGD("VirtualSingleVerSyncDBInterface::GetMaxTimestamp time = %" PRIu64, stamp);
213 }
214
GetMetaData(const Key & key,Value & value) const215 int VirtualRelationalVerSyncDBInterface::GetMetaData(const Key &key, Value &value) const
216 {
217 auto iter = metadata_.find(key);
218 if (iter != metadata_.end()) {
219 value = iter->second;
220 return E_OK;
221 }
222 return -E_NOT_FOUND;
223 }
224
PutMetaData(const Key & key,const Value & value)225 int VirtualRelationalVerSyncDBInterface::PutMetaData(const Key &key, const Value &value)
226 {
227 metadata_[key] = value;
228 return E_OK;
229 }
230
DeleteMetaData(const std::vector<Key> & keys)231 int VirtualRelationalVerSyncDBInterface::DeleteMetaData(const std::vector<Key> &keys)
232 {
233 for (const auto &key : keys) {
234 (void)metadata_.erase(key);
235 }
236 return E_OK;
237 }
238
DeleteMetaDataByPrefixKey(const Key & keyPrefix) const239 int VirtualRelationalVerSyncDBInterface::DeleteMetaDataByPrefixKey(const Key &keyPrefix) const
240 {
241 size_t prefixKeySize = keyPrefix.size();
242 for (auto iter = metadata_.begin();iter != metadata_.end();) {
243 if (prefixKeySize <= iter->first.size() &&
244 keyPrefix == Key(iter->first.begin(), std::next(iter->first.begin(), prefixKeySize))) {
245 iter = metadata_.erase(iter);
246 } else {
247 ++iter;
248 }
249 }
250 return E_OK;
251 }
252
GetAllMetaKeys(std::vector<Key> & keys) const253 int VirtualRelationalVerSyncDBInterface::GetAllMetaKeys(std::vector<Key> &keys) const
254 {
255 for (const auto &iter : metadata_) {
256 keys.push_back(iter.first);
257 }
258 LOGD("GetAllMetaKeys size %zu", keys.size());
259 return E_OK;
260 }
261
GetDbProperties() const262 const RelationalDBProperties &VirtualRelationalVerSyncDBInterface::GetDbProperties() const
263 {
264 return rdbProperties_;
265 }
266
SetLocalFieldInfo(const std::vector<FieldInfo> & localFieldInfo)267 void VirtualRelationalVerSyncDBInterface::SetLocalFieldInfo(const std::vector<FieldInfo> &localFieldInfo)
268 {
269 localFieldInfo_.clear();
270 localFieldInfo_ = localFieldInfo;
271 }
272
GetAllSyncData(const std::string & tableName,std::vector<VirtualRowData> & data)273 int VirtualRelationalVerSyncDBInterface::GetAllSyncData(const std::string &tableName,
274 std::vector<VirtualRowData> &data)
275 {
276 if (syncData_.find(tableName) == syncData_.end()) {
277 return -E_NOT_FOUND;
278 }
279 for (const auto &entry : syncData_[tableName]) {
280 if (entry.second.logInfo.flag != DataItem::DELETE_FLAG) {
281 data.push_back(entry.second);
282 }
283 }
284 return E_OK;
285 }
286
GetVirtualSyncData(const std::string & tableName,const std::string & hashKey,VirtualRowData & data)287 int VirtualRelationalVerSyncDBInterface::GetVirtualSyncData(const std::string &tableName,
288 const std::string &hashKey, VirtualRowData &data)
289 {
290 if (syncData_.find(tableName) == syncData_.end()) {
291 return -E_NOT_FOUND;
292 }
293 if (syncData_.find(hashKey) == syncData_.end()) {
294 return -E_NOT_FOUND;
295 }
296 data = syncData_[tableName][hashKey];
297 return E_OK;
298 }
299
EraseSyncData(const std::string & tableName)300 void VirtualRelationalVerSyncDBInterface::EraseSyncData(const std::string &tableName)
301 {
302 if (syncData_.find(tableName) == syncData_.end()) {
303 return;
304 }
305 syncData_.erase(tableName);
306 }
307
CreateDistributedDeviceTable(const std::string & device,const RelationalSyncStrategy & syncStrategy)308 int VirtualRelationalVerSyncDBInterface::CreateDistributedDeviceTable(const std::string &device,
309 const RelationalSyncStrategy &syncStrategy)
310 {
311 return permitCreateDistributedTable_ ? E_OK : -E_NOT_SUPPORT;
312 }
313
RegisterSchemaChangedCallback(const std::function<void ()> & onSchemaChanged)314 int VirtualRelationalVerSyncDBInterface::RegisterSchemaChangedCallback(const std::function<void()> &onSchemaChanged)
315 {
316 return E_OK;
317 }
318
SetTableInfo(const TableInfo & tableInfo)319 void VirtualRelationalVerSyncDBInterface::SetTableInfo(const TableInfo &tableInfo)
320 {
321 schemaObj_.AddRelationalTable(tableInfo);
322 }
323
GetMaxTimestamp(const std::string & tableName,Timestamp & timestamp) const324 int VirtualRelationalVerSyncDBInterface::GetMaxTimestamp(const std::string &tableName, Timestamp ×tamp) const
325 {
326 (void)tableName;
327 timestamp = 0;
328 return E_OK;
329 }
330
ExecuteQuery(const PreparedStmt & prepStmt,size_t packetSize,RelationalRowDataSet & data,ContinueToken & token) const331 int VirtualRelationalVerSyncDBInterface::ExecuteQuery(const PreparedStmt &prepStmt, size_t packetSize,
332 RelationalRowDataSet &data, ContinueToken &token) const
333 {
334 return E_OK;
335 }
336
SaveRemoteDeviceSchema(const std::string & deviceId,const std::string & remoteSchema,uint8_t type)337 int VirtualRelationalVerSyncDBInterface::SaveRemoteDeviceSchema(const std::string &deviceId,
338 const std::string &remoteSchema, uint8_t type)
339 {
340 return E_OK;
341 }
342
GetRemoteDeviceSchema(const std::string & deviceId,RelationalSchemaObject & schemaObj)343 int VirtualRelationalVerSyncDBInterface::GetRemoteDeviceSchema(const std::string &deviceId,
344 RelationalSchemaObject &schemaObj)
345 {
346 return E_OK;
347 }
348
SetPermitCreateDistributedTable(bool permitCreateDistributedTable)349 void VirtualRelationalVerSyncDBInterface::SetPermitCreateDistributedTable(bool permitCreateDistributedTable)
350 {
351 permitCreateDistributedTable_ = permitCreateDistributedTable;
352 }
353
PutDataValue(const std::string & fieldName,const DataValue & value) const354 void ObjectData::PutDataValue(const std::string &fieldName, const DataValue &value) const
355 {
356 fieldData[fieldName] = value;
357 }
358
GetDataValue(const std::string & fieldName,DataValue & value) const359 int ObjectData::GetDataValue(const std::string &fieldName, DataValue &value) const
360 {
361 if (fieldData.find(fieldName) == fieldData.end()) {
362 return -E_NOT_FOUND;
363 }
364 value = fieldData[fieldName];
365 return E_OK;
366 }
367
GetSecurityOption(SecurityOption & option) const368 int VirtualRelationalVerSyncDBInterface::GetSecurityOption(SecurityOption &option) const
369 {
370 return RuntimeContext::GetInstance()->GetSecurityOption("", option);
371 }
372
ReleaseRemoteQueryContinueToken(ContinueToken & token) const373 void VirtualRelationalVerSyncDBInterface::ReleaseRemoteQueryContinueToken(ContinueToken &token) const
374 {
375 auto remoteToken = static_cast<RelationalRemoteQueryContinueToken *>(token);
376 delete remoteToken;
377 remoteToken = nullptr;
378 token = nullptr;
379 }
380 }
381 #endif
382