Cherry pick 3.3 (1013-1017)

conf/nebula-graphd.conf.default

@@ -90,3 +90,9 @@
 ########## experimental feature ##########
 # if use experimental features
 --enable_experimental_feature=false
+
+# if use toss feature, only work if enable_experimental_feature is true
+--enable_toss=false
+
+# if use balance data feature, only work if enable_experimental_feature is true
+--enable_data_balance=true

conf/nebula-graphd.conf.production

@@ -90,6 +90,12 @@
 # if use experimental features
 --enable_experimental_feature=false
 
+# if use toss feature, only work if enable_experimental_feature is true
+--enable_toss=false
+
+# if use balance data feature, only work if enable_experimental_feature is true
+--enable_data_balance=true
+
 ########## session ##########
 # Maximum number of sessions that can be created per IP and per user
 --max_sessions_per_ip_per_user=300

conf/nebula-standalone.conf.default

@@ -92,6 +92,12 @@
 # if use experimental features
 --enable_experimental_feature=false
 
+# if use toss feature, only work if enable_experimental_feature is true
+--enable_toss=false
+
+# if use balance data feature, only work if enable_experimental_feature is true
+--enable_data_balance=true
+
 ######### Raft #########
 # Raft election timeout
 --raft_heartbeat_interval_secs=30

src/common/hdfs/HdfsCommandHelper.cpp

@@ -24,6 +24,9 @@ Status HdfsCommandHelper::ls(const std::string& hdfsHost,
     auto result = proc.wait();
     if (!result.exited()) {
       return Status::Error("Failed to ls hdfs");
+    } else if (result.exitStatus() != 0) {
+      LOG(INFO) << "Failed to ls: " << result.str();
+      return Status::Error("Failed to ls hdfs, errno: %d", result.exitStatus());
     } else {
       return Status::OK();
     }
@@ -47,6 +50,9 @@ Status HdfsCommandHelper::copyToLocal(const std::string& hdfsHost,
     auto result = proc.wait();
     if (!result.exited()) {
       return Status::Error("Failed to download from hdfs");
+    } else if (result.exitStatus() != 0) {
+      LOG(INFO) << "Failed to download: " << result.str();
+      return Status::Error("Failed to download from hdfs, errno: %d", result.exitStatus());
     } else {
       return Status::OK();
     }

src/graph/executor/mutate/DeleteExecutor.cpp

@@ -200,7 +200,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;
+  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
   return qctx()
       ->getStorageClient()
       ->deleteEdges(param, std::move(edgeKeys))

src/graph/executor/mutate/InsertExecutor.cpp

@@ -54,7 +54,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;
+  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
   return qctx()
       ->getStorageClient()
       ->addEdges(param,

src/graph/executor/mutate/UpdateExecutor.cpp

@@ -98,7 +98,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;
+  param.useExperimentalFeature = FLAGS_enable_experimental_feature && FLAGS_enable_toss;
   return qctx()
       ->getStorageClient()
       ->updateEdge(param,

src/graph/service/GraphFlags.cpp

@@ -70,6 +70,8 @@ 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");
 DEFINE_int32(max_sessions_per_ip_per_user,
@@ -103,3 +105,5 @@ DEFINE_uint32(
     gc_worker_size,
     0,
     "Background garbage clean workers, default number is 0 which means using hardware core size.");
+
+DEFINE_bool(graph_use_vertex_key, false, "whether allow insert or query the vertex key");

src/graph/service/GraphFlags.h

@@ -54,6 +54,8 @@ 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);
 DECLARE_string(client_white_list);
@@ -65,4 +67,7 @@ DECLARE_int32(max_job_size);
 
 DECLARE_bool(enable_async_gc);
 DECLARE_uint32(gc_worker_size);
+
+DECLARE_bool(graph_use_vertex_key);
+
 #endif  // GRAPH_GRAPHFLAGS_H_

src/graph/validator/AdminJobValidator.cpp

@@ -6,14 +6,15 @@
 #include "graph/validator/AdminJobValidator.h"
 
 #include "graph/planner/plan/Admin.h"
+#include "graph/service/GraphFlags.h"
 
 namespace nebula {
 namespace graph {
 
 Status AdminJobValidator::validateImpl() {
   if (sentence_->getJobType() == meta::cpp2::JobType::DATA_BALANCE ||
       sentence_->getJobType() == meta::cpp2::JobType::ZONE_BALANCE) {
-    if (!FLAGS_enable_experimental_feature) {
+    if (!(FLAGS_enable_experimental_feature && FLAGS_enable_data_balance)) {
       return Status::SemanticError("Data balance not support");
     }
   }

src/graph/validator/MutateValidator.cpp

@@ -46,6 +46,9 @@ Status InsertVerticesValidator::check() {
   }
 
   auto tagItems = sentence->tagItems();
+  if (!FLAGS_graph_use_vertex_key && tagItems.empty()) {
+    return Status::SemanticError("Insert vertex is forbidden, please speicify the tag");
+  }
 
   schemas_.reserve(tagItems.size());
 
@@ -206,7 +209,8 @@ 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 ? rows_.size() : rows_.size() * 2;
+  auto size =
+      FLAGS_enable_experimental_feature && FLAGS_enable_toss ? rows_.size() : rows_.size() * 2;
   edges_.reserve(size);
 
   size_t fieldNum = schema_->getNumFields();
@@ -291,7 +295,7 @@ Status InsertEdgesValidator::prepareEdges() {
     edge.key_ref() = key;
     edge.props_ref() = std::move(entirePropValues);
     edges_.emplace_back(edge);
-    if (!FLAGS_enable_experimental_feature) {
+    if (!(FLAGS_enable_experimental_feature && FLAGS_enable_toss)) {
       // inbound
       key.src_ref() = dstId;
       key.dst_ref() = srcId;
@@ -826,7 +830,7 @@ Status UpdateEdgeValidator::toPlan() {
                                    {},
                                    condition_,
                                    {});
-  if (FLAGS_enable_experimental_feature) {
+  if ((FLAGS_enable_experimental_feature && FLAGS_enable_toss)) {
     root_ = outNode;
     tail_ = root_;
   } else {

src/graph/validator/test/YieldValidatorTest.cpp

@@ -162,15 +162,11 @@ TEST_F(YieldValidatorTest, TypeCastTest) {
   }
   {
     std::string query = "YIELD (int)\"123abc\"";
-    auto result = checkResult(query);
-    EXPECT_EQ(std::string(result.message()),
-              "SemanticError: `(INT)\"123abc\"' is not a valid expression ");
+    EXPECT_TRUE(checkResult(query));
   }
   {
     std::string query = "YIELD (int)\"abc123\"";
-    auto result = checkResult(query);
-    EXPECT_EQ(std::string(result.message()),
-              "SemanticError: `(INT)\"abc123\"' is not a valid expression ");
+    EXPECT_TRUE(checkResult(query));
   }
   {
     std::string query = "YIELD (doublE)\"123\"";
@@ -182,9 +178,7 @@ TEST_F(YieldValidatorTest, TypeCastTest) {
   }
   {
     std::string query = "YIELD (doublE)\".a123\"";
-    auto result = checkResult(query);
-    EXPECT_EQ(std::string(result.message()),
-              "SemanticError: `(FLOAT)\".a123\"' is not a valid expression ");
+    EXPECT_TRUE(checkResult(query));
   }
   {
     std::string query = "YIELD (STRING)1.23";
@@ -200,15 +194,15 @@ TEST_F(YieldValidatorTest, TypeCastTest) {
   }
   {
     std::string query = "YIELD (BOOL)123";
-    EXPECT_FALSE(checkResult(query, expected_));
+    EXPECT_TRUE(checkResult(query, expected_));
   }
   {
     std::string query = "YIELD (BOOL)0";
-    EXPECT_FALSE(checkResult(query, expected_));
+    EXPECT_TRUE(checkResult(query, expected_));
   }
   {
     std::string query = "YIELD (BOOL)\"12\"";
-    EXPECT_FALSE(checkResult(query, expected_));
+    EXPECT_TRUE(checkResult(query, expected_));
   }
   {
     std::string query = "YIELD (MAP)(\"12\")";

src/graph/visitor/DeduceTypeVisitor.cpp

@@ -249,13 +249,7 @@ void DeduceTypeVisitor::visit(TypeCastingExpression *expr) {
     status_ = Status::SemanticError(out.str());
     return;
   }
-  QueryExpressionContext ctx(nullptr);
-  auto val = expr->eval(ctx(nullptr));
-  if (val.isNull()) {
-    status_ = Status::SemanticError("`%s' is not a valid expression ", expr->toString().c_str());
-    return;
-  }
-  type_ = val.type();
+  type_ = expr->type();
   status_ = Status::OK();
 }
 
@@ -486,7 +480,21 @@ void DeduceTypeVisitor::visit(FunctionCallExpression *expr) {
 void DeduceTypeVisitor::visit(AggregateExpression *expr) {
   expr->arg()->accept(this);
   if (!ok()) return;
-  type_ = Value::Type::__EMPTY__;
+  auto func = expr->name();
+  std::transform(func.begin(), func.end(), func.begin(), ::toupper);
+  if ("COUNT" == func) {
+    type_ = Value::Type::INT;
+  } else if ("COLLECT" == func) {
+    type_ = Value::Type::LIST;
+  } else if ("COLLECT_SET" == func) {
+    type_ = Value::Type::SET;
+  } else if ("AVG" == func || "SUM" == func) {
+    type_ = Value::Type::FLOAT;
+  } else if ("MAX" == func || "MIN" == func) {
+    // Keep same with arg's type
+  } else {
+    type_ = Value::Type::__EMPTY__;
+  }
 }
 
 void DeduceTypeVisitor::visit(UUIDExpression *) {

src/storage/admin/RebuildIndexTask.cpp

@@ -17,6 +17,17 @@ const int32_t kReserveNum = 1024 * 4;
 bool RebuildIndexTask::check() {
   return env_->kvstore_ != nullptr;
 }
+void RebuildIndexTask::finish(nebula::cpp2::ErrorCode rc) {
+  if (changedSpaceGuard_) {
+    auto space = *ctx_.parameters_.space_id_ref();
+    for (auto it = env_->rebuildIndexGuard_->begin(); it != env_->rebuildIndexGuard_->end(); ++it) {
+      if (std::get<0>(it->first) == space) {
+        env_->rebuildIndexGuard_->insert_or_assign(it->first, IndexState::FINISHED);
+      }
+    }
+  }
+  AdminTask::finish(rc);
+}
 
 RebuildIndexTask::RebuildIndexTask(StorageEnv* env, TaskContext&& ctx)
     : AdminTask(env, std::move(ctx)) {
@@ -71,6 +82,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<AdminSubTask>> RebuildIndexTask::ge
   for (const auto& part : parts) {
     env_->rebuildIndexGuard_->insert_or_assign(std::make_tuple(space_, part), IndexState::STARTING);
     TaskFunction task = std::bind(&RebuildIndexTask::invoke, this, space_, part, items);
+    changedSpaceGuard_ = true;
     tasks.emplace_back(std::move(task));
   }
   return tasks;

src/storage/admin/RebuildIndexTask.h

@@ -23,6 +23,8 @@ using IndexItems = std::vector<std::shared_ptr<meta::cpp2::IndexItem>>;
  */
 class RebuildIndexTask : public AdminTask {
  public:
+  using AdminTask::finish;
+
   RebuildIndexTask(StorageEnv* env, TaskContext&& ctx);
 
   ~RebuildIndexTask() {
@@ -31,6 +33,8 @@ class RebuildIndexTask : public AdminTask {
 
   bool check() override;
 
+  void finish(nebula::cpp2::ErrorCode rc) override;
+
   /**
    * @brief Generate subtasks for rebuilding index.
    *
@@ -71,6 +75,7 @@ class RebuildIndexTask : public AdminTask {
 
  protected:
   GraphSpaceID space_;
+  bool changedSpaceGuard_{false};
 };
 
 }  // namespace storage

src/storage/exec/GetPropNode.h

@@ -21,15 +21,17 @@ class GetTagPropNode : public QueryNode<VertexID> {
                  std::vector<TagNode*> tagNodes,
                  nebula::DataSet* resultDataSet,
                  Expression* filter,
-                 std::size_t limit)
+                 std::size_t limit,
+                 TagContext* tagContext)
       : context_(context),
         tagNodes_(std::move(tagNodes)),
         resultDataSet_(resultDataSet),
         expCtx_(filter == nullptr
                     ? nullptr
                     : new StorageExpressionContext(context->vIdLen(), context->isIntId())),
         filter_(filter),
-        limit_(limit) {
+        limit_(limit),
+        tagContext_(tagContext) {
     name_ = "GetTagPropNode";
   }
 
@@ -69,7 +71,36 @@ class GetTagPropNode : public QueryNode<VertexID> {
         } else if (!iter->valid()) {
           return nebula::cpp2::ErrorCode::SUCCEEDED;
         }
-        // if has any tag, will emplace a row with vId
+
+        bool hasValidTag = false;
+        for (; iter->valid(); iter->next()) {
+          // check if tag schema exists
+          auto key = iter->key();
+          auto tagId = NebulaKeyUtils::getTagId(context_->vIdLen(), key);
+          auto schemaIter = tagContext_->schemas_.find(tagId);
+          if (schemaIter == tagContext_->schemas_.end()) {
+            continue;
+          }
+          // check if ttl expired
+          auto schemas = &(schemaIter->second);
+          RowReaderWrapper reader;
+          reader.reset(*schemas, iter->val());
+          if (!reader) {
+            continue;
+          }
+          auto ttl = QueryUtils::getTagTTLInfo(tagContext_, tagId);
+          if (ttl.has_value() &&
+              CommonUtils::checkDataExpiredForTTL(
+                  schemas->back().get(), reader.get(), ttl.value().first, ttl.value().second)) {
+            continue;
+          }
+          hasValidTag = true;
+          break;
+        }
+        if (!hasValidTag) {
+          return nebula::cpp2::ErrorCode::SUCCEEDED;
+        }
+        // if has any valid tag, will emplace a row with vId
       }
     }
 
@@ -131,6 +162,7 @@ class GetTagPropNode : public QueryNode<VertexID> {
   std::unique_ptr<StorageExpressionContext> expCtx_{nullptr};
   Expression* filter_{nullptr};
   const std::size_t limit_{std::numeric_limits<std::size_t>::max()};
+  TagContext* tagContext_;
 };
 
 class GetEdgePropNode : public QueryNode<cpp2::EdgeKey> {

src/storage/query/GetPropProcessor.cpp

@@ -208,7 +208,7 @@ StoragePlan<VertexID> GetPropProcessor::buildTagPlan(RuntimeContext* context,
     plan.addNode(std::move(tag));
   }
   auto output = std::make_unique<GetTagPropNode>(
-      context, tags, result, filter_ == nullptr ? nullptr : filter_->clone(), limit_);
+      context, tags, result, filter_ == nullptr ? nullptr : filter_->clone(), limit_, &tagContext_);
   for (auto* tag : tags) {
     output->addDependency(tag);
   }

tests/tck/features/delete/DeleteTag.IntVid.feature

@@ -161,7 +161,6 @@ Feature: Delete int vid of tag
       | id |
     Then drop the used space
 
-  @wtf
   Scenario: delete int vid multiple vertex one tag
     Given an empty graph
     And load "nba_int_vid" csv data to a new space