Fix memory leak, remove toss gflag

src/graph/executor/mutate/DeleteExecutor.cpp

@@ -215,7 +215,7 @@ folly::Future<Status> DeleteEdgesExecutor::deleteEdges() {
   auto plan = qctx()->plan();
   StorageClient::CommonRequestParam param(
       spaceId, qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
-  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
+  param.useExperimentalFeature = false;
   return qctx()
       ->getStorageClient()
       ->deleteEdges(param, std::move(edgeKeys))

src/graph/executor/mutate/InsertExecutor.cpp

@@ -61,7 +61,7 @@ folly::Future<Status> InsertEdgesExecutor::insertEdges() {
   auto plan = qctx()->plan();
   StorageClient::CommonRequestParam param(
       ieNode->getSpace(), qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
-  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
+  param.useExperimentalFeature = false;
   return qctx()
       ->getStorageClient()
       ->addEdges(param,

src/graph/executor/mutate/UpdateExecutor.cpp

@@ -105,7 +105,7 @@ folly::Future<Status> UpdateEdgeExecutor::execute() {
   auto plan = qctx()->plan();
   StorageClient::CommonRequestParam param(
       ueNode->getSpaceId(), qctx()->rctx()->session()->id(), plan->id(), plan->isProfileEnabled());
-  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
+  param.useExperimentalFeature = false;
   return qctx()
       ->getStorageClient()
       ->updateEdge(param,

src/graph/service/GraphFlags.cpp

@@ -70,7 +70,6 @@ DEFINE_bool(disable_octal_escape_char,
             " in next version to ensure compatibility with cypher.");
 
 DEFINE_bool(enable_experimental_feature, false, "Whether to enable experimental feature");
-DEFINE_bool(enable_toss, false, "Whether to enable toss feature");
 DEFINE_bool(enable_data_balance, true, "Whether to enable data balance feature");
 
 DEFINE_int32(num_rows_to_check_memory, 1024, "number rows to check memory");

src/graph/service/GraphFlags.h

@@ -54,7 +54,6 @@ DECLARE_bool(optimize_appendvertice);
 DECLARE_int64(max_allowed_connections);
 
 DECLARE_bool(enable_experimental_feature);
-DECLARE_bool(enable_toss);
 DECLARE_bool(enable_data_balance);
 
 DECLARE_bool(enable_client_white_list);

src/graph/validator/MutateValidator.cpp

@@ -211,8 +211,7 @@ Status InsertEdgesValidator::check() {
 // Check validity of vertices data.
 // Check edge key type, check properties value, fill to NewEdge structure.
 Status InsertEdgesValidator::prepareEdges() {
-  auto size =
-      FLAGS_enable_experimental_feature && FLAGS_enable_toss ? rows_.size() : rows_.size() * 2;
+  auto size = rows_.size() * 2;
   edges_.reserve(size);
 
   size_t fieldNum = schema_->getNumFields();
@@ -297,7 +296,7 @@ Status InsertEdgesValidator::prepareEdges() {
     edge.key_ref() = key;
     edge.props_ref() = std::move(entirePropValues);
     edges_.emplace_back(edge);
-    if (!(FLAGS_enable_experimental_feature && FLAGS_enable_toss)) {
+    {
       // inbound
       key.src_ref() = dstId;
       key.dst_ref() = srcId;
@@ -892,26 +891,21 @@ Status UpdateEdgeValidator::toPlan() {
                                    {},
                                    condition_,
                                    {});
-  if ((FLAGS_enable_experimental_feature && FLAGS_enable_toss)) {
-    root_ = outNode;
-    tail_ = root_;
-  } else {
-    auto *inNode = UpdateEdge::make(qctx_,
-                                    outNode,
-                                    spaceId_,
-                                    std::move(name_),
-                                    std::move(dstId_),
-                                    std::move(srcId_),
-                                    -edgeType_,
-                                    rank_,
-                                    insertable_,
-                                    std::move(updatedProps_),
-                                    std::move(returnProps_),
-                                    std::move(condition_),
-                                    std::move(yieldColNames_));
-    root_ = inNode;
-    tail_ = outNode;
-  }
+  auto *inNode = UpdateEdge::make(qctx_,
+                                  outNode,
+                                  spaceId_,
+                                  std::move(name_),
+                                  std::move(dstId_),
+                                  std::move(srcId_),
+                                  -edgeType_,
+                                  rank_,
+                                  insertable_,
+                                  std::move(updatedProps_),
+                                  std::move(returnProps_),
+                                  std::move(condition_),
+                                  std::move(yieldColNames_));
+  root_ = inNode;
+  tail_ = outNode;
   return Status::OK();
 }
 

src/meta/processors/index/FTIndexProcessor.cpp

@@ -173,13 +173,6 @@ void DropFTIndexProcessor::process(const cpp2::DropFTIndexReq& req) {
     return;
   }
 
-  auto batchHolder = std::make_unique<kvstore::BatchHolder>();
-  batchHolder->remove(std::move(indexKey));
-  auto timeInMilliSec = time::WallClock::fastNowInMilliSec();
-  LastUpdateTimeMan::update(batchHolder.get(), timeInMilliSec);
-  auto batch = encodeBatchValue(std::move(batchHolder)->getBatch());
-  doBatchOperation(std::move(batch));
-
   const auto& serviceKey = MetaKeyUtils::serviceKey(cpp2::ExternalServiceType::ELASTICSEARCH);
   auto getRet = doGet(serviceKey);
   if (!nebula::ok(getRet)) {
@@ -213,6 +206,13 @@ void DropFTIndexProcessor::process(const cpp2::DropFTIndexReq& req) {
     onFinished();
     return;
   }
+
+  auto batchHolder = std::make_unique<kvstore::BatchHolder>();
+  batchHolder->remove(std::move(indexKey));
+  auto timeInMilliSec = time::WallClock::fastNowInMilliSec();
+  LastUpdateTimeMan::update(batchHolder.get(), timeInMilliSec);
+  auto batch = encodeBatchValue(std::move(batchHolder)->getBatch());
+  doBatchOperation(std::move(batch));
 }
 
 void ListFTIndexesProcessor::process(const cpp2::ListFTIndexesReq&) {

tests/tck/conftest.py

@@ -86,7 +86,7 @@ def running_or_queue_row(row):
 
 
 def wait_all_jobs_finished(sess, jobs=[]):
-    times = 4 * get_running_jobs(sess)
+    times = 5 * get_running_jobs(sess)
     while jobs and times > 0:
         jobs = [job for job in jobs if not is_job_finished(sess, job)]
         time.sleep(1)