1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "query_sync_object.h"
17
18 #include "cloud/cloud_db_constant.h"
19 #include "db_common.h"
20 #include "db_errno.h"
21 #include "log_print.h"
22 #include "version.h"
23
24 namespace DistributedDB {
25 namespace {
26 const std::string MAGIC = "remote query";
27 // Max value size of each QueryObjNode, current is In & NotIn predicate which is 128
28 const int MAX_VALUE_SIZE = 128;
29 const int MAX_QUERY_NODE_SIZE = 256;
30
SerializeDataObjNode(Parcel & parcel,const QueryObjNode & objNode)31 int SerializeDataObjNode(Parcel &parcel, const QueryObjNode &objNode)
32 {
33 if (objNode.operFlag == QueryObjType::OPER_ILLEGAL) {
34 return -E_INVALID_QUERY_FORMAT;
35 }
36 (void)parcel.WriteUInt32(static_cast<uint32_t>(objNode.operFlag));
37 parcel.EightByteAlign();
38 (void)parcel.WriteString(objNode.fieldName);
39 (void)parcel.WriteInt(static_cast<int32_t>(objNode.type));
40 (void)parcel.WriteUInt32(objNode.fieldValue.size());
41
42 for (const FieldValue &value : objNode.fieldValue) {
43 (void)parcel.WriteString(value.stringValue);
44
45 // string may not closely arranged continuously
46 // longValue is maximum length in union
47 (void)parcel.WriteInt64(value.longValue);
48 }
49 if (parcel.IsError()) {
50 return -E_INVALID_ARGS;
51 }
52 return E_OK;
53 }
54
DeSerializeDataObjNode(Parcel & parcel,QueryObjNode & objNode)55 int DeSerializeDataObjNode(Parcel &parcel, QueryObjNode &objNode)
56 {
57 uint32_t readOperFlag = 0;
58 (void)parcel.ReadUInt32(readOperFlag);
59 objNode.operFlag = static_cast<QueryObjType>(readOperFlag);
60 parcel.EightByteAlign();
61
62 (void)parcel.ReadString(objNode.fieldName);
63
64 int readInt = -1;
65 (void)parcel.ReadInt(readInt);
66 objNode.type = static_cast<QueryValueType>(readInt);
67
68 uint32_t valueSize = 0;
69 (void)parcel.ReadUInt32(valueSize);
70 if (parcel.IsError() || valueSize > MAX_VALUE_SIZE) {
71 return -E_INVALID_ARGS;
72 }
73
74 for (size_t i = 0; i < valueSize; i++) {
75 FieldValue value;
76 (void)parcel.ReadString(value.stringValue);
77
78 (void)parcel.ReadInt64(value.longValue);
79 if (parcel.IsError()) {
80 return -E_INVALID_ARGS;
81 }
82 objNode.fieldValue.push_back(value);
83 }
84 return E_OK;
85 }
86 }
87
QuerySyncObject()88 QuerySyncObject::QuerySyncObject()
89 {}
90
QuerySyncObject(const std::list<QueryObjNode> & queryObjNodes,const std::vector<uint8_t> & prefixKey,const std::set<Key> & keys)91 QuerySyncObject::QuerySyncObject(const std::list<QueryObjNode> &queryObjNodes, const std::vector<uint8_t> &prefixKey,
92 const std::set<Key> &keys)
93 : QueryObject(queryObjNodes, prefixKey, keys)
94 {}
95
QuerySyncObject(const Query & query)96 QuerySyncObject::QuerySyncObject(const Query &query)
97 : QueryObject(query)
98 {}
99
QuerySyncObject(const DistributedDB::QueryExpression & expression)100 QuerySyncObject::QuerySyncObject(const DistributedDB::QueryExpression &expression)
101 : QueryObject(expression)
102 {}
103
~QuerySyncObject()104 QuerySyncObject::~QuerySyncObject()
105 {}
106
GetVersion() const107 uint32_t QuerySyncObject::GetVersion() const
108 {
109 uint32_t version = QUERY_SYNC_OBJECT_VERSION_0;
110 if (isTableNameSpecified_ || !keys_.empty()) {
111 version = QUERY_SYNC_OBJECT_VERSION_1;
112 }
113 return version;
114 }
115
GetObjContext(ObjContext & objContext) const116 int QuerySyncObject::GetObjContext(ObjContext &objContext) const
117 {
118 if (!isValid_) {
119 return -E_INVALID_QUERY_FORMAT;
120 }
121 objContext.version = GetVersion();
122 objContext.prefixKey.assign(prefixKey_.begin(), prefixKey_.end());
123 objContext.suggestIndex = suggestIndex_;
124 objContext.queryObjNodes = queryObjNodes_;
125 return E_OK;
126 }
127
CalculateIdentifyLen() const128 uint32_t QuerySyncObject::CalculateIdentifyLen() const
129 {
130 uint64_t len = Parcel::GetVectorCharLen(prefixKey_);
131 for (const QueryObjNode &node : queryObjNodes_) {
132 if (node.operFlag == QueryObjType::LIMIT || node.operFlag == QueryObjType::ORDERBY ||
133 node.operFlag == QueryObjType::SUGGEST_INDEX) {
134 continue;
135 }
136 // operFlag and valueType is int
137 len += Parcel::GetUInt32Len() + Parcel::GetIntLen() + Parcel::GetStringLen(node.fieldName);
138 for (const FieldValue &value : node.fieldValue) {
139 len += Parcel::GetStringLen(value.stringValue) + Parcel::GetInt64Len();
140 }
141 }
142
143 // QUERY_SYNC_OBJECT_VERSION_1 added.
144 len += isTableNameSpecified_ ? Parcel::GetStringLen(tableName_) : 0;
145 for (const auto &key : keys_) {
146 len += Parcel::GetVectorCharLen(key);
147 } // QUERY_SYNC_OBJECT_VERSION_1 end.
148 return len;
149 }
150
GetIdentify() const151 std::string QuerySyncObject::GetIdentify() const
152 {
153 if (!isValid_) {
154 return std::string();
155 }
156 if (!identify_.empty()) {
157 return identify_;
158 }
159 // suggestionIndex is local attribute, do not need to be propagated to remote
160 uint64_t len = CalculateIdentifyLen();
161 std::vector<uint8_t> buff(len, 0); // It will affect the hash result, the default value cannot be modified
162 Parcel parcel(buff.data(), len);
163
164 // The order needs to be consistent, otherwise it will affect the hash result
165 (void)parcel.WriteVectorChar(prefixKey_);
166 for (const QueryObjNode &node : queryObjNodes_) {
167 if (node.operFlag == QueryObjType::LIMIT || node.operFlag == QueryObjType::ORDERBY ||
168 node.operFlag == QueryObjType::SUGGEST_INDEX) {
169 continue;
170 }
171 (void)parcel.WriteUInt32(static_cast<uint32_t>(node.operFlag));
172 (void)parcel.WriteInt(static_cast<int32_t>(node.type));
173 (void)parcel.WriteString(node.fieldName);
174 for (const FieldValue &value : node.fieldValue) {
175 (void)parcel.WriteInt64(value.longValue);
176 (void)parcel.WriteString(value.stringValue);
177 }
178 }
179
180 // QUERY_SYNC_OBJECT_VERSION_1 added.
181 if (isTableNameSpecified_) {
182 (void)parcel.WriteString(tableName_);
183 }
184 for (const auto &key : keys_) {
185 (void)parcel.WriteVectorChar(key);
186 } // QUERY_SYNC_OBJECT_VERSION_1 end.
187
188 std::vector<uint8_t> hashBuff;
189 if (parcel.IsError() || DBCommon::CalcValueHash(buff, hashBuff) != E_OK) {
190 return std::string();
191 }
192 identify_ = DBCommon::VectorToHexString(hashBuff);
193 return identify_;
194 }
195
CalculateParcelLen(uint32_t softWareVersion) const196 uint32_t QuerySyncObject::CalculateParcelLen(uint32_t softWareVersion) const
197 {
198 if (softWareVersion == SOFTWARE_VERSION_CURRENT) {
199 return CalculateLen();
200 }
201 LOGE("current not support!");
202 return 0;
203 }
204
SerializeData(Parcel & parcel,uint32_t softWareVersion)205 int QuerySyncObject::SerializeData(Parcel &parcel, uint32_t softWareVersion)
206 {
207 ObjContext context;
208 int errCode = GetObjContext(context);
209 if (errCode != E_OK) {
210 return errCode;
211 }
212 (void)parcel.WriteString(MAGIC);
213 (void)parcel.WriteUInt32(context.version);
214 (void)parcel.WriteVectorChar(context.prefixKey);
215 (void)parcel.WriteString(context.suggestIndex);
216 (void)parcel.WriteUInt32(context.queryObjNodes.size());
217 parcel.EightByteAlign();
218 if (parcel.IsError()) {
219 return -E_INVALID_ARGS;
220 }
221 for (const QueryObjNode &node : context.queryObjNodes) {
222 errCode = SerializeDataObjNode(parcel, node);
223 if (errCode != E_OK) {
224 return errCode;
225 }
226 }
227
228 // QUERY_SYNC_OBJECT_VERSION_1 added.
229 if (context.version >= QUERY_SYNC_OBJECT_VERSION_1) {
230 (void)parcel.WriteUInt32(static_cast<uint32_t>(isTableNameSpecified_));
231 if (isTableNameSpecified_) {
232 (void)parcel.WriteString(tableName_);
233 }
234 (void)parcel.WriteUInt32(keys_.size());
235 for (const auto &key : keys_) {
236 (void)parcel.WriteVectorChar(key);
237 }
238 } // QUERY_SYNC_OBJECT_VERSION_1 end.
239 parcel.EightByteAlign();
240 if (parcel.IsError()) { // parcel almost success
241 return -E_INVALID_ARGS;
242 }
243 return E_OK;
244 }
245
SetCloudGid(const std::vector<std::string> & cloudGid)246 void QuerySyncObject::SetCloudGid(const std::vector<std::string> &cloudGid)
247 {
248 for (size_t i = 0; i < cloudGid.size(); i+= MAX_VALUE_SIZE) {
249 size_t end = std::min(i + MAX_VALUE_SIZE, cloudGid.size());
250 if (!queryObjNodes_.empty()) {
251 QueryObjNode operateNode;
252 operateNode.operFlag = QueryObjType::OR;
253 operateNode.type = QueryValueType::VALUE_TYPE_NULL;
254 queryObjNodes_.emplace_back(operateNode);
255 }
256
257 QueryObjNode objNode;
258 objNode.operFlag = QueryObjType::IN;
259 objNode.fieldName = CloudDbConstant::GID_FIELD;
260 objNode.type = QueryValueType::VALUE_TYPE_STRING;
261 std::vector<std::string> subCloudGid(cloudGid.begin() + i, cloudGid.begin() + end);
262 for (const auto &gid : subCloudGid) {
263 if (gid.empty()) {
264 continue;
265 }
266 FieldValue fieldValue;
267 fieldValue.stringValue = gid;
268 objNode.fieldValue.emplace_back(fieldValue);
269 }
270 queryObjNodes_.emplace_back(objNode);
271 }
272 }
273
274 namespace {
DeSerializeVersion1Data(uint32_t version,Parcel & parcel,std::string & tableName,std::set<Key> & keys)275 int DeSerializeVersion1Data(uint32_t version, Parcel &parcel, std::string &tableName, std::set<Key> &keys)
276 {
277 if (version >= QUERY_SYNC_OBJECT_VERSION_1) {
278 uint32_t isTblNameExist = 0;
279 (void)parcel.ReadUInt32(isTblNameExist);
280 if (isTblNameExist) {
281 (void)parcel.ReadString(tableName);
282 }
283 uint32_t keysSize = 0;
284 (void)parcel.ReadUInt32(keysSize);
285 if (keysSize > DBConstant::MAX_INKEYS_SIZE) {
286 return -E_PARSE_FAIL;
287 }
288 for (uint32_t i = 0; i < keysSize; ++i) {
289 Key key;
290 (void)parcel.ReadVector(key);
291 keys.emplace(key);
292 }
293 }
294 return E_OK;
295 }
296 }
297
DeSerializeData(Parcel & parcel,QuerySyncObject & queryObj)298 int QuerySyncObject::DeSerializeData(Parcel &parcel, QuerySyncObject &queryObj)
299 {
300 std::string magic;
301 (void)parcel.ReadString(magic);
302 if (magic != MAGIC) {
303 return -E_INVALID_ARGS;
304 }
305
306 ObjContext context;
307 (void)parcel.ReadUInt32(context.version);
308 if (context.version > QUERY_SYNC_OBJECT_VERSION_CURRENT) {
309 LOGE("Parcel version and deserialize version not matched! ver=%u", context.version);
310 return -E_VERSION_NOT_SUPPORT;
311 }
312
313 (void)parcel.ReadVectorChar(context.prefixKey);
314 (void)parcel.ReadString(context.suggestIndex);
315
316 uint32_t nodesSize = 0;
317 (void)parcel.ReadUInt32(nodesSize);
318 parcel.EightByteAlign();
319 // Due to historical reasons, the limit of query node size was incorrectly set to MAX_QUERY_NODE_SIZE + 1
320 if (parcel.IsError() || nodesSize > MAX_QUERY_NODE_SIZE + 1) { // almost success
321 return -E_INVALID_ARGS;
322 }
323 for (size_t i = 0; i < nodesSize; i++) {
324 QueryObjNode node;
325 int errCode = DeSerializeDataObjNode(parcel, node);
326 if (errCode != E_OK) {
327 return errCode;
328 }
329 context.queryObjNodes.emplace_back(node);
330 }
331
332 // QUERY_SYNC_OBJECT_VERSION_1 added.
333 std::string tableName;
334 std::set<Key> keys;
335 int errCode = DeSerializeVersion1Data(context.version, parcel, tableName, keys);
336 if (errCode != E_OK) {
337 return errCode;
338 } // QUERY_SYNC_OBJECT_VERSION_1 end.
339
340 if (parcel.IsError()) { // almost success
341 return -E_INVALID_ARGS;
342 }
343 queryObj = QuerySyncObject(context.queryObjNodes, context.prefixKey, keys);
344 if (!tableName.empty()) {
345 queryObj.SetTableName(tableName);
346 }
347 return E_OK;
348 }
349
CalculateLen() const350 uint32_t QuerySyncObject::CalculateLen() const
351 {
352 uint64_t len = Parcel::GetStringLen(MAGIC);
353 len += Parcel::GetUInt32Len(); // version
354 len += Parcel::GetVectorCharLen(prefixKey_);
355 len += Parcel::GetStringLen(suggestIndex_);
356 len += Parcel::GetUInt32Len(); // nodes size
357 len = Parcel::GetEightByteAlign(len);
358 for (const QueryObjNode &node : queryObjNodes_) {
359 if (node.operFlag == QueryObjType::OPER_ILLEGAL) {
360 LOGE("contain illegal operator for query sync!");
361 return 0;
362 }
363 // operflag, fieldName, query value type, value size, union max size, string value
364 len += Parcel::GetUInt32Len();
365 len = Parcel::GetEightByteAlign(len);
366 len += Parcel::GetStringLen(node.fieldName) +
367 Parcel::GetIntLen() + Parcel::GetUInt32Len();
368 for (size_t i = 0; i < node.fieldValue.size(); i++) {
369 len += Parcel::GetInt64Len() + Parcel::GetStringLen(node.fieldValue[i].stringValue);
370 }
371 }
372
373 // QUERY_SYNC_OBJECT_VERSION_1 added.
374 len += Parcel::GetUInt32Len(); // whether the table name exists.
375 if (isTableNameSpecified_) {
376 len += Parcel::GetStringLen(tableName_);
377 }
378 len += Parcel::GetUInt32Len(); // size of keys_
379 for (const auto &key : keys_) {
380 len += Parcel::GetVectorCharLen(key);
381 } // QUERY_SYNC_OBJECT_VERSION_1 end.
382
383 len = Parcel::GetEightByteAlign(len);
384 if (len > INT32_MAX) {
385 return 0;
386 }
387 return static_cast<uint32_t>(len);
388 }
389
GetRelationTableName() const390 std::string QuerySyncObject::GetRelationTableName() const
391 {
392 if (!isTableNameSpecified_) {
393 return {};
394 }
395 return tableName_;
396 }
397
GetRelationTableNames() const398 std::vector<std::string> QuerySyncObject::GetRelationTableNames() const
399 {
400 return tables_;
401 }
402
GetValidStatus() const403 int QuerySyncObject::GetValidStatus() const
404 {
405 return validStatus;
406 }
407
IsContainQueryNodes() const408 bool QuerySyncObject::IsContainQueryNodes() const
409 {
410 return !queryObjNodes_.empty();
411 }
412
IsInValueOutOfLimit() const413 bool QuerySyncObject::IsInValueOutOfLimit() const
414 {
415 for (const auto &queryObjNode : queryObjNodes_) {
416 if ((queryObjNode.operFlag == QueryObjType::IN) &&
417 (queryObjNode.fieldValue.size() > DBConstant::MAX_IN_COUNT)) {
418 return false;
419 }
420 }
421 return true;
422 }
423
GetQuerySyncObject(const DistributedDB::Query & query)424 std::vector<QuerySyncObject> QuerySyncObject::GetQuerySyncObject(const DistributedDB::Query &query)
425 {
426 std::vector<QuerySyncObject> res;
427 const auto &expressions = QueryObject::GetQueryExpressions(query);
428 for (const auto &item : expressions) {
429 res.push_back(QuerySyncObject(item));
430 }
431 return res;
432 }
433
ParserQueryNodes(const Bytes & bytes,std::vector<QueryNode> & queryNodes)434 int QuerySyncObject::ParserQueryNodes(const Bytes &bytes, std::vector<QueryNode> &queryNodes)
435 {
436 QuerySyncObject tmp;
437 Bytes parcelBytes = bytes;
438 Parcel parcel(parcelBytes.data(), parcelBytes.size());
439 int errCode = DeSerializeData(parcel, tmp);
440 if (errCode != E_OK) {
441 return errCode;
442 }
443 for (const auto &objNode: tmp.queryObjNodes_) {
444 QueryNode node;
445 errCode = TransformToQueryNode(objNode, node);
446 if (errCode != E_OK) {
447 return errCode;
448 }
449 queryNodes.push_back(std::move(node));
450 }
451 return E_OK;
452 }
453
TransformToQueryNode(const QueryObjNode & objNode,QueryNode & node)454 int QuerySyncObject::TransformToQueryNode(const QueryObjNode &objNode, QueryNode &node)
455 {
456 int errCode = TransformValueToType(objNode, node.fieldValue);
457 if (errCode != E_OK) {
458 LOGE("[Query] transform value to type failed %d", errCode);
459 return errCode;
460 }
461 node.fieldName = objNode.fieldName;
462 return TransformNodeType(objNode, node);
463 }
464
TransformValueToType(const QueryObjNode & objNode,std::vector<Type> & types)465 int QuerySyncObject::TransformValueToType(const QueryObjNode &objNode, std::vector<Type> &types)
466 {
467 for (const auto &value: objNode.fieldValue) {
468 switch (objNode.type) {
469 case QueryValueType::VALUE_TYPE_STRING:
470 types.emplace_back(value.stringValue);
471 break;
472 case QueryValueType::VALUE_TYPE_BOOL:
473 types.emplace_back(value.boolValue);
474 break;
475 case QueryValueType::VALUE_TYPE_NULL:
476 types.emplace_back(Nil());
477 break;
478 case QueryValueType::VALUE_TYPE_INTEGER:
479 case QueryValueType::VALUE_TYPE_LONG:
480 types.emplace_back(static_cast<int64_t>(value.integerValue));
481 break;
482 case QueryValueType::VALUE_TYPE_DOUBLE:
483 types.emplace_back(value.doubleValue);
484 break;
485 case QueryValueType::VALUE_TYPE_INVALID:
486 return -E_INVALID_ARGS;
487 }
488 }
489 return E_OK;
490 }
491
TransformNodeType(const QueryObjNode & objNode,QueryNode & node)492 int QuerySyncObject::TransformNodeType(const QueryObjNode &objNode, QueryNode &node)
493 {
494 int errCode = E_OK;
495 switch (objNode.operFlag) {
496 case QueryObjType::IN:
497 node.type = QueryNodeType::IN;
498 break;
499 case QueryObjType::OR:
500 node.type = QueryNodeType::OR;
501 break;
502 case QueryObjType::AND:
503 node.type = QueryNodeType::AND;
504 break;
505 case QueryObjType::EQUALTO:
506 node.type = QueryNodeType::EQUAL_TO;
507 break;
508 case QueryObjType::BEGIN_GROUP:
509 node.type = QueryNodeType::BEGIN_GROUP;
510 break;
511 case QueryObjType::END_GROUP:
512 node.type = QueryNodeType::END_GROUP;
513 break;
514 case QueryObjType::IN_KEYS:
515 node.fieldName = CloudDbConstant::CLOUD_KV_FIELD_KEY;
516 node.type = QueryNodeType::IN;
517 break;
518 default:
519 LOGE("[Query] not support type %d", static_cast<int>(objNode.operFlag));
520 errCode = -E_NOT_SUPPORT;
521 node.type = QueryNodeType::ILLEGAL;
522 }
523 return errCode;
524 }
525
GetQuerySyncObjectFromGroup(int64_t groupId,QuerySyncObject & obj)526 int QuerySyncObject::GetQuerySyncObjectFromGroup(int64_t groupId, QuerySyncObject &obj)
527 {
528 obj = *this;
529 if (groupNum_ <= 1) {
530 return E_OK;
531 }
532 // find the begin group node
533 bool isFindBeginGroup = false;
534 int64_t beginGroupIndex = 0;
535 for (auto iter = obj.queryObjNodes_.begin(); iter != obj.queryObjNodes_.end();) {
536 if ((*iter).operFlag != QueryObjType::BEGIN_GROUP) {
537 // eraes the node which is before the begin group node
538 iter = obj.queryObjNodes_.erase(iter);
539 continue;
540 } else if (beginGroupIndex != groupId) {
541 // eraes the node which is before the begin group node
542 iter = obj.queryObjNodes_.erase(iter);
543 beginGroupIndex++;
544 continue;
545 } else {
546 isFindBeginGroup = true;
547 break;
548 }
549 }
550 if (!isFindBeginGroup) {
551 LOGE("can not find the begin group node, groupid %u", groupId);
552 return -E_INVALID_ARGS;
553 }
554
555 // find the end group node
556 bool isFindEndGroup = false;
557 for (auto iter = obj.queryObjNodes_.begin(); iter != obj.queryObjNodes_.end();) {
558 if (isFindEndGroup) {
559 // eraes the node which is behind the end group node
560 iter = obj.queryObjNodes_.erase(iter);
561 continue;
562 } else if ((*iter).operFlag == QueryObjType::END_GROUP) {
563 isFindEndGroup = true;
564 }
565 ++iter;
566 }
567 if (!isFindEndGroup) {
568 LOGE("can not find the end group node, groupid %u", groupId);
569 return -E_INVALID_ARGS;
570 }
571 return E_OK;
572 }
573 } // namespace DistributedDB