add gflags to control max_edge_returned_per_vertex (vesoft-inc#1221)

* add gflags to control max edge returned from one vertex

* rename gflags

* fix logic error, from max vertice to max edge

* while add new UT, replace an old one by mistake... fix it

* change default max edge to INT_MAX

src/storage/QueryBaseProcessor.cpp

@@ -8,6 +8,7 @@
 
 DEFINE_int32(max_handlers_per_req, 10, "The max handlers used to handle one request");
 DEFINE_int32(min_vertices_per_bucket, 3, "The min vertices number in one bucket");
+DEFINE_int32(max_edge_returned_per_vertex, INT_MAX, "Max edge number returnred searching vertex");
 
 namespace nebula {
 namespace storage {

src/storage/QueryBaseProcessor.inl

@@ -11,6 +11,7 @@
 
 DECLARE_int32(max_handlers_per_req);
 DECLARE_int32(min_vertices_per_bucket);
+DECLARE_int32(max_edge_returned_per_vertex);
 
 namespace nebula {
 namespace storage {
@@ -393,7 +394,8 @@ kvstore::ResultCode QueryBaseProcessor<REQ, RESP>::collectEdgeProps(
  EdgeRanking lastRank = -1;
  VertexID lastDstId = 0;
  bool firstLoop = true;
- for (; iter->valid(); iter->next()) {
+ int cnt = 0;
+ for (; iter->valid() && cnt < FLAGS_max_edge_returned_per_vertex; iter->next()) {
  auto key = iter->key();
  auto val = iter->val();
  auto rank = NebulaKeyUtils::getRank(key);
@@ -447,6 +449,7 @@ kvstore::ResultCode QueryBaseProcessor<REQ, RESP>::collectEdgeProps(
  }
  }
  proc(reader.get(), key, props);
+ ++cnt;
  if (firstLoop) {
  firstLoop = false;
  }

src/storage/test/QueryBoundTest.cpp

@@ -474,6 +474,33 @@ TEST(QueryBoundTest, MultiEdgeQueryTest) {
  checkResponse(resp, 30, 12, 10001, 7, true);
 }
 
+TEST(QueryBoundTest, MaxEdgesReturenedTest) {
+ int old_max_edge_returned = FLAGS_max_edge_returned_per_vertex;
+ FLAGS_max_edge_returned_per_vertex = 5;
+ fs::TempDir rootPath("/tmp/QueryBoundTest.XXXXXX");
+ LOG(INFO) << "Prepare meta...";
+ std::unique_ptr<kvstore::KVStore> kv = TestUtils::initKV(rootPath.path());
+
+ auto schemaMan = TestUtils::mockSchemaMan();
+ mockData(kv.get());
+
+ cpp2::GetNeighborsRequest req;
+ std::vector<EdgeType> et = {101};
+ buildRequest(req, et);
+
+ LOG(INFO) << "Test QueryOutBoundRequest...";
+ auto executor = std::make_unique<folly::CPUThreadPoolExecutor>(3);
+ auto* processor = QueryBoundProcessor::instance(kv.get(), schemaMan.get(), nullptr,
+ executor.get());
+ auto f = processor->getFuture();
+ processor->process(req);
+ auto resp = std::move(f).get();
+
+ LOG(INFO) << "Check the results...";
+ checkResponse(resp, 30, 12, 10001, FLAGS_max_edge_returned_per_vertex, true);
+ FLAGS_max_edge_returned_per_vertex = old_max_edge_returned;
+}
+
 } // namespace storage
 } // namespace nebula