Skip to content

Commit 20df766

Browse files
darionyaphetdarionyaphet
committed
fix test case
1 parent bf3ba6e commit 20df766

17 files changed

+426
-317
lines changed

‎src/meta/processors/parts/CreateSpaceAsProcessor.cpp

+8-5
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,6 @@ void CreateSpaceAsProcessor::process(const cpp2::CreateSpaceAsReq &req) {
4747
}
4848

4949
std::vector<kvstore::KV> data;
50-
5150
auto newSpaceData =
5251
makeNewSpaceData(nebula::value(oldSpaceId), nebula::value(newSpaceId), newSpaceName);
5352
if (nebula::ok(newSpaceData)) {
@@ -123,9 +122,10 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
123122
return nebula::error(partPrefix);
124123
}
125124
auto iter = nebula::value(partPrefix).get();
126-
for (; iter->valid(); iter->next()) {
125+
while (iter->valid()) {
127126
auto partId = MetaKeyUtils::parsePartKeyPartId(iter->key());
128127
data.emplace_back(MetaKeyUtils::partKey(newSpaceId, partId), iter->val());
128+
iter->next();
129129
}
130130
return data;
131131
}
@@ -145,7 +145,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
145145

146146
std::vector<kvstore::KV> data;
147147
auto iter = nebula::value(tagPrefix).get();
148-
for (; iter->valid(); iter->next()) {
148+
while (iter->valid()) {
149149
auto val = iter->val();
150150

151151
auto tagId = MetaKeyUtils::parseTagId(iter->key());
@@ -157,6 +157,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
157157
auto tagVer = MetaKeyUtils::parseTagVersion(iter->key());
158158
auto key = MetaKeyUtils::schemaTagKey(newSpaceId, tagId, tagVer);
159159
data.emplace_back(std::move(key), val.str());
160+
iter->next();
160161
}
161162
return data;
162163
}
@@ -176,7 +177,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
176177

177178
std::vector<kvstore::KV> data;
178179
auto iter = nebula::value(edgePrefix).get();
179-
for (; iter->valid(); iter->next()) {
180+
while (iter->valid()) {
180181
auto val = iter->val();
181182

182183
auto edgeType = MetaKeyUtils::parseEdgeType(iter->key());
@@ -188,6 +189,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
188189
auto ver = MetaKeyUtils::parseEdgeVersion(iter->key());
189190
auto key = MetaKeyUtils::schemaEdgeKey(newSpaceId, edgeType, ver);
190191
data.emplace_back(std::move(key), val.str());
192+
iter->next();
191193
}
192194
return data;
193195
}
@@ -207,7 +209,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
207209

208210
std::vector<kvstore::KV> data;
209211
auto iter = nebula::value(indexPrefix).get();
210-
for (; iter->valid(); iter->next()) {
212+
while (iter->valid()) {
211213
auto val = iter->val();
212214

213215
auto indexId = MetaKeyUtils::parseIndexesKeyIndexID(iter->key());
@@ -219,6 +221,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
219221
std::string(reinterpret_cast<const char *>(&indexId), sizeof(indexId)));
220222

221223
data.emplace_back(MetaKeyUtils::indexKey(newSpaceId, indexId), MetaKeyUtils::indexVal(idxItem));
224+
iter->next();
222225
}
223226
return data;
224227
}

‎src/meta/processors/parts/CreateSpaceProcessor.cpp

+111-127
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
6161
// storage
6262
properties.set_partition_num(partitionNum);
6363
}
64+
6465
if (replicaFactor == 0) {
6566
replicaFactor = FLAGS_default_replica_factor;
6667
if (replicaFactor <= 0) {
@@ -73,12 +74,14 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
7374
// storage
7475
properties.set_replica_factor(replicaFactor);
7576
}
77+
7678
if (vidSize == 0) {
7779
LOG(ERROR) << "Create Space Failed : vid_size is illegal: " << vidSize;
7880
handleErrorCode(nebula::cpp2::ErrorCode::E_INVALID_PARM);
7981
onFinished();
8082
return;
8183
}
84+
8285
if (vidType != nebula::cpp2::PropertyType::INT64 &&
8386
vidType != nebula::cpp2::PropertyType::FIXED_STRING) {
8487
LOG(ERROR) << "Create Space Failed : vid_type is illegal: "
@@ -87,6 +90,7 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
8790
onFinished();
8891
return;
8992
}
93+
9094
if (vidType == nebula::cpp2::PropertyType::INT64 && vidSize != 8) {
9195
LOG(ERROR) << "Create Space Failed : vid_size should be 8 if vid type is integer: " << vidSize;
9296
handleErrorCode(nebula::cpp2::ErrorCode::E_INVALID_PARM);
@@ -105,148 +109,139 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
105109

106110
auto spaceId = nebula::value(idRet);
107111
std::vector<kvstore::KV> data;
108-
data.emplace_back(MetaKeyUtils::indexSpaceKey(spaceName),
109-
std::string(reinterpret_cast<const char*>(&spaceId), sizeof(spaceId)));
110-
data.emplace_back(MetaKeyUtils::spaceKey(spaceId), MetaKeyUtils::spaceVal(properties));
111-
112+
std::vector<::std::string> zones;
112113
nebula::cpp2::ErrorCode code = nebula::cpp2::ErrorCode::SUCCEEDED;
113-
if (!properties.get_zone_names().empty()) {
114-
auto zones = properties.get_zone_names();
115-
for (auto& zone : zones) {
116-
auto zoneKey = MetaKeyUtils::zoneKey(zone);
117-
auto ret = doGet(zoneKey);
118-
if (!nebula::ok(ret)) {
119-
auto retCode = nebula::error(ret);
120-
if (retCode == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
121-
code = nebula::cpp2::ErrorCode::E_ZONE_NOT_FOUND;
122-
}
123-
LOG(ERROR) << " Get Zone Name: " << zone << " failed.";
124-
break;
125-
}
126-
}
127-
128-
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
129-
LOG(ERROR) << "Create space failed";
114+
if (properties.get_zone_names().empty()) {
115+
const auto& zonePrefix = MetaKeyUtils::zonePrefix();
116+
auto zoneIterRet = doPrefix(zonePrefix);
117+
if (!nebula::ok(zoneIterRet)) {
118+
code = nebula::error(zoneIterRet);
119+
LOG(ERROR) << "Get zones failed, error: " << apache::thrift::util::enumNameSafe(code);
130120
handleErrorCode(code);
131121
onFinished();
132122
return;
133123
}
134124

135-
int32_t zoneNum = zones.size();
136-
if (replicaFactor > zoneNum) {
137-
LOG(ERROR) << "Replication number should less than or equal to zone number.";
138-
handleErrorCode(nebula::cpp2::ErrorCode::E_INVALID_PARM);
139-
onFinished();
140-
return;
125+
auto zoneIter = nebula::value(zoneIterRet).get();
126+
while (zoneIter->valid()) {
127+
auto zoneName = MetaKeyUtils::parseZoneName(zoneIter->key());
128+
zones.emplace_back(std::move(zoneName));
129+
zoneIter->next();
141130
}
142131

143-
auto hostLoadingRet = getHostLoading();
144-
if (!nebula::ok(hostLoadingRet)) {
145-
LOG(ERROR) << "Get host loading failed.";
146-
auto retCode = nebula::error(hostLoadingRet);
147-
if (retCode != nebula::cpp2::ErrorCode::E_LEADER_CHANGED) {
148-
retCode = nebula::cpp2::ErrorCode::E_INVALID_PARM;
132+
properties.set_zone_names(zones);
133+
} else {
134+
zones = properties.get_zone_names();
135+
}
136+
137+
data.emplace_back(MetaKeyUtils::indexSpaceKey(spaceName),
138+
std::string(reinterpret_cast<const char*>(&spaceId), sizeof(spaceId)));
139+
data.emplace_back(MetaKeyUtils::spaceKey(spaceId), MetaKeyUtils::spaceVal(properties));
140+
for (auto& zone : zones) {
141+
auto zoneKey = MetaKeyUtils::zoneKey(zone);
142+
auto ret = doGet(zoneKey);
143+
if (!nebula::ok(ret)) {
144+
auto retCode = nebula::error(ret);
145+
if (retCode == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
146+
code = nebula::cpp2::ErrorCode::E_ZONE_NOT_FOUND;
149147
}
150-
handleErrorCode(retCode);
151-
onFinished();
152-
return;
148+
LOG(ERROR) << " Get Zone Name: " << zone << " failed.";
149+
break;
153150
}
151+
}
154152

155-
hostLoading_ = std::move(nebula::value(hostLoadingRet));
156-
std::unordered_map<std::string, Hosts> zoneHosts;
157-
for (auto& zone : zones) {
158-
auto zoneKey = MetaKeyUtils::zoneKey(zone);
159-
auto zoneValueRet = doGet(std::move(zoneKey));
160-
if (!nebula::ok(zoneValueRet)) {
161-
code = nebula::error(zoneValueRet);
162-
if (code == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
163-
code = nebula::cpp2::ErrorCode::E_ZONE_NOT_FOUND;
164-
}
165-
LOG(ERROR) << "Get zone " << zone << " failed.";
166-
break;
167-
}
153+
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
154+
LOG(ERROR) << "Create space failed";
155+
handleErrorCode(code);
156+
onFinished();
157+
return;
158+
}
168159

169-
auto hosts = MetaKeyUtils::parseZoneHosts(std::move(nebula::value(zoneValueRet)));
170-
for (auto& host : hosts) {
171-
auto hostIter = hostLoading_.find(host);
172-
if (hostIter == hostLoading_.end()) {
173-
hostLoading_[host] = 0;
174-
zoneLoading_[zone] += 0;
175-
} else {
176-
zoneLoading_[zone] += hostIter->second;
177-
}
178-
}
179-
zoneHosts[zone] = std::move(hosts);
180-
}
160+
int32_t zoneNum = zones.size();
161+
if (replicaFactor > zoneNum) {
162+
LOG(ERROR) << "Replication number should less than or equal to zone number.";
163+
LOG(ERROR) << "Replication number: " << replicaFactor << ", Zones size: " << zones.size();
164+
handleErrorCode(nebula::cpp2::ErrorCode::E_INVALID_PARM);
165+
onFinished();
166+
return;
167+
}
181168

182-
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
183-
LOG(ERROR) << "Create space failed";
184-
handleErrorCode(code);
185-
onFinished();
186-
return;
169+
auto hostLoadingRet = getHostLoading();
170+
if (!nebula::ok(hostLoadingRet)) {
171+
LOG(ERROR) << "Get host loading failed.";
172+
auto retCode = nebula::error(hostLoadingRet);
173+
if (retCode != nebula::cpp2::ErrorCode::E_LEADER_CHANGED) {
174+
retCode = nebula::cpp2::ErrorCode::E_INVALID_PARM;
187175
}
176+
handleErrorCode(retCode);
177+
onFinished();
178+
return;
179+
}
188180

189-
for (auto partId = 1; partId <= partitionNum; partId++) {
190-
auto pickedZonesRet = pickLightLoadZones(replicaFactor);
191-
if (!pickedZonesRet.ok()) {
192-
LOG(ERROR) << "Pick zone failed.";
193-
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
194-
break;
195-
}
196-
197-
auto pickedZones = std::move(pickedZonesRet).value();
198-
auto partHostsRet = pickHostsWithZone(pickedZones, zoneHosts);
199-
if (!partHostsRet.ok()) {
200-
LOG(ERROR) << "Pick hosts with zone failed.";
201-
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
202-
break;
181+
hostLoading_ = std::move(nebula::value(hostLoadingRet));
182+
std::unordered_map<std::string, Hosts> zoneHosts;
183+
for (auto& zone : zones) {
184+
auto zoneKey = MetaKeyUtils::zoneKey(zone);
185+
auto zoneValueRet = doGet(std::move(zoneKey));
186+
if (!nebula::ok(zoneValueRet)) {
187+
code = nebula::error(zoneValueRet);
188+
if (code == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
189+
code = nebula::cpp2::ErrorCode::E_ZONE_NOT_FOUND;
203190
}
191+
LOG(ERROR) << "Get zone " << zone << " failed.";
192+
break;
193+
}
204194

205-
auto partHosts = std::move(partHostsRet).value();
206-
if (partHosts.empty()) {
207-
LOG(ERROR) << "Pick hosts is empty.";
208-
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
209-
break;
195+
auto hosts = MetaKeyUtils::parseZoneHosts(std::move(nebula::value(zoneValueRet)));
196+
for (auto& host : hosts) {
197+
auto hostIter = hostLoading_.find(host);
198+
if (hostIter == hostLoading_.end()) {
199+
hostLoading_[host] = 0;
200+
zoneLoading_[zone] += 0;
201+
} else {
202+
zoneLoading_[zone] += hostIter->second;
210203
}
204+
}
205+
zoneHosts[zone] = std::move(hosts);
206+
}
211207

212-
std::stringstream ss;
213-
for (const auto& host : partHosts) {
214-
ss << host << ", ";
215-
}
208+
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
209+
LOG(ERROR) << "Create space failed";
210+
handleErrorCode(code);
211+
onFinished();
212+
return;
213+
}
216214

217-
VLOG(3) << "Space " << spaceId << " part " << partId << " hosts " << ss.str();
218-
data.emplace_back(MetaKeyUtils::partKey(spaceId, partId), MetaKeyUtils::partVal(partHosts));
219-
}
220-
} else {
221-
auto hostsRet = ActiveHostsMan::getActiveHosts(kvstore_);
222-
if (!nebula::ok(hostsRet)) {
223-
auto retCode = nebula::error(hostsRet);
224-
LOG(ERROR) << "Create Space Failed when get active host, error "
225-
<< apache::thrift::util::enumNameSafe(retCode);
226-
handleErrorCode(retCode);
227-
onFinished();
228-
return;
215+
for (auto partId = 1; partId <= partitionNum; partId++) {
216+
auto pickedZonesRet = pickLightLoadZones(replicaFactor);
217+
if (!pickedZonesRet.ok()) {
218+
LOG(ERROR) << "Pick zone failed.";
219+
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
220+
break;
229221
}
230-
auto hosts = std::move(nebula::value(hostsRet));
231-
if (hosts.empty()) {
232-
LOG(ERROR) << "Create Space Failed : No Hosts!";
233-
handleErrorCode(nebula::cpp2::ErrorCode::E_NO_HOSTS);
234-
onFinished();
235-
return;
222+
223+
auto pickedZones = std::move(pickedZonesRet).value();
224+
auto partHostsRet = pickHostsWithZone(pickedZones, zoneHosts);
225+
if (!partHostsRet.ok()) {
226+
LOG(ERROR) << "Pick hosts with zone failed.";
227+
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
228+
break;
236229
}
237230

238-
if ((int32_t)hosts.size() < replicaFactor) {
239-
LOG(ERROR) << "Not enough hosts existed for replica " << replicaFactor << ", hosts num "
240-
<< hosts.size();
241-
handleErrorCode(nebula::cpp2::ErrorCode::E_INVALID_PARM);
242-
onFinished();
243-
return;
231+
auto partHosts = std::move(partHostsRet).value();
232+
if (partHosts.empty()) {
233+
LOG(ERROR) << "Pick hosts is empty.";
234+
code = nebula::cpp2::ErrorCode::E_INVALID_PARM;
235+
break;
244236
}
245237

246-
for (auto partId = 1; partId <= partitionNum; partId++) {
247-
auto partHosts = pickHosts(partId, hosts, replicaFactor);
248-
data.emplace_back(MetaKeyUtils::partKey(spaceId, partId), MetaKeyUtils::partVal(partHosts));
238+
std::stringstream ss;
239+
for (const auto& host : partHosts) {
240+
ss << host << ", ";
249241
}
242+
243+
VLOG(3) << "Space " << spaceId << " part " << partId << " hosts " << ss.str();
244+
data.emplace_back(MetaKeyUtils::partKey(spaceId, partId), MetaKeyUtils::partVal(partHosts));
250245
}
251246

252247
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
@@ -261,17 +256,6 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
261256
LOG(INFO) << "Create space " << spaceName;
262257
}
263258

264-
Hosts CreateSpaceProcessor::pickHosts(PartitionID partId,
265-
const Hosts& hosts,
266-
int32_t replicaFactor) {
267-
auto startIndex = partId;
268-
Hosts pickedHosts;
269-
for (int32_t i = 0; i < replicaFactor; i++) {
270-
pickedHosts.emplace_back(toThriftHost(hosts[startIndex++ % hosts.size()]));
271-
}
272-
return pickedHosts;
273-
}
274-
275259
ErrorOr<nebula::cpp2::ErrorCode, std::unordered_map<HostAddr, int32_t>>
276260
CreateSpaceProcessor::getHostLoading() {
277261
const auto& prefix = MetaKeyUtils::partPrefix();

‎src/meta/processors/parts/CreateSpaceProcessor.h

-2
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,6 @@ class CreateSpaceProcessor : public BaseProcessor<cpp2::ExecResp> {
2525
explicit CreateSpaceProcessor(kvstore::KVStore* kvstore)
2626
: BaseProcessor<cpp2::ExecResp>(kvstore) {}
2727

28-
Hosts pickHosts(PartitionID partId, const Hosts& hosts, int32_t replicaFactor);
29-
3028
// Get the host with the least load in the zone
3129
StatusOr<Hosts> pickHostsWithZone(const std::vector<std::string>& zones,
3230
const std::unordered_map<std::string, Hosts>& zoneHosts);

0 commit comments

Comments
 (0)