Merge branch 'unstable' into explainsql-dot

apache · Jun 15, 2024 · bf278bd · bf278bd
2 parents 8156705 + 8f1d2ad
commit bf278bd
Show file tree

Hide file tree

Showing 24 changed files with 162 additions and 218 deletions.
diff --git a/src/cluster/cluster.cc b/src/cluster/cluster.cc
@@ -363,7 +363,7 @@ Status Cluster::ImportSlot(redis::Connection *conn, int slot, int state) {
 
 Status Cluster::GetClusterInfo(std::string *cluster_infos) {
   if (version_ < 0) {
-    return {Status::ClusterDown, errClusterNoInitialized};
+    return {Status::RedisClusterDown, errClusterNoInitialized};
   }
 
   cluster_infos->clear();
@@ -421,7 +421,7 @@ Status Cluster::GetClusterInfo(std::string *cluster_infos) {
 //          ... continued until done
 Status Cluster::GetSlotsInfo(std::vector<SlotInfo> *slots_infos) {
   if (version_ < 0) {
-    return {Status::ClusterDown, errClusterNoInitialized};
+    return {Status::RedisClusterDown, errClusterNoInitialized};
   }
 
   slots_infos->clear();
@@ -464,7 +464,7 @@ SlotInfo Cluster::genSlotNodeInfo(int start, int end, const std::shared_ptr<Clus
 // $version $connected $slot_range
 Status Cluster::GetClusterNodes(std::string *nodes_str) {
   if (version_ < 0) {
-    return {Status::ClusterDown, errClusterNoInitialized};
+    return {Status::RedisClusterDown, errClusterNoInitialized};
   }
 
   *nodes_str = genNodesDescription();
@@ -473,7 +473,7 @@ Status Cluster::GetClusterNodes(std::string *nodes_str) {
 
 StatusOr<std::string> Cluster::GetReplicas(const std::string &node_id) {
   if (version_ < 0) {
-    return {Status::ClusterDown, errClusterNoInitialized};
+    return {Status::RedisClusterDown, errClusterNoInitialized};
   }
 
   auto item = nodes_.find(node_id);
@@ -821,26 +821,26 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
     int cur_slot = GetSlotIdFromKey(cmd_tokens[i]);
     if (slot == -1) slot = cur_slot;
     if (slot != cur_slot) {
-      return {Status::RedisExecErr, "CROSSSLOT Attempted to access keys that don't hash to the same slot"};
+      return {Status::RedisCrossSlot, "Attempted to access keys that don't hash to the same slot"};
     }
   }
   if (slot == -1) return Status::OK();
 
   if (slots_nodes_[slot] == nullptr) {
-    return {Status::ClusterDown, "CLUSTERDOWN Hash slot not served"};
+    return {Status::RedisClusterDown, "Hash slot not served"};
   }
 
   if (myself_ && myself_ == slots_nodes_[slot]) {
     // We use central controller to manage the topology of the cluster.
     // Server can't change the topology directly, so we record the migrated slots
     // to move the requests of the migrated slots to the destination node.
     if (migrated_slots_.count(slot) > 0) {  // I'm not serving the migrated slot
-      return {Status::RedisExecErr, fmt::format("MOVED {} {}", slot, migrated_slots_[slot])};
+      return {Status::RedisMoved, fmt::format("{} {}", slot, migrated_slots_[slot])};
     }
     // To keep data consistency, slot will be forbidden write while sending the last incremental data.
     // During this phase, the requests of the migrating slot has to be rejected.
     if ((attributes->flags & redis::kCmdWrite) && IsWriteForbiddenSlot(slot)) {
-      return {Status::RedisExecErr, "TRYAGAIN Can't write to slot being migrated which is in write forbidden phase"};
+      return {Status::RedisTryAgain, "Can't write to slot being migrated which is in write forbidden phase"};
     }
 
     return Status::OK();  // I'm serving this slot
@@ -868,8 +868,7 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
     return Status::OK();  // My master is serving this slot
   }
 
-  return {Status::RedisExecErr,
-          fmt::format("MOVED {} {}:{}", slot, slots_nodes_[slot]->host, slots_nodes_[slot]->port)};
+  return {Status::RedisMoved, fmt::format("{} {}:{}", slot, slots_nodes_[slot]->host, slots_nodes_[slot]->port)};
 }
 
 // Only HARD mode is meaningful to the Kvrocks cluster,

diff --git a/src/cluster/cluster_defs.h b/src/cluster/cluster_defs.h
@@ -36,7 +36,7 @@ inline constexpr const char *errSlotOutOfRange = "Slot is out of range";
 inline constexpr const char *errInvalidClusterVersion = "Invalid cluster version";
 inline constexpr const char *errSlotOverlapped = "Slot distribution is overlapped";
 inline constexpr const char *errNoMasterNode = "The node isn't a master";
-inline constexpr const char *errClusterNoInitialized = "CLUSTERDOWN The cluster is not initialized";
+inline constexpr const char *errClusterNoInitialized = "The cluster is not initialized";
 inline constexpr const char *errInvalidClusterNodeInfo = "Invalid cluster nodes info";
 inline constexpr const char *errInvalidImportState = "Invalid import state";
 

diff --git a/src/cluster/replication.cc b/src/cluster/replication.cc
@@ -1000,15 +1000,18 @@ Status ReplicationThread::parseWriteBatch(const std::string &batch_string) {
 }
 
 bool ReplicationThread::isRestoringError(std::string_view err) {
-  return err == std::string(RESP_PREFIX_ERROR) + redis::errRestoringBackup;
+  // err doesn't contain the CRLF, so cannot use redis::Error here.
+  return err == RESP_PREFIX_ERROR + redis::StatusToRedisErrorMsg({Status::RedisLoading, redis::errRestoringBackup});
 }
 
 bool ReplicationThread::isWrongPsyncNum(std::string_view err) {
-  return err == std::string(RESP_PREFIX_ERROR) + redis::errWrongNumArguments;
+  // err doesn't contain the CRLF, so cannot use redis::Error here.
+  return err == RESP_PREFIX_ERROR + redis::StatusToRedisErrorMsg({Status::NotOK, redis::errWrongNumArguments});
 }
 
 bool ReplicationThread::isUnknownOption(std::string_view err) {
-  return err == fmt::format("{}ERR {}", RESP_PREFIX_ERROR, redis::errUnknownOption);
+  // err doesn't contain the CRLF, so cannot use redis::Error here.
+  return err == RESP_PREFIX_ERROR + redis::StatusToRedisErrorMsg({Status::NotOK, redis::errUnknownOption});
 }
 
 rocksdb::Status WriteBatchHandler::PutCF(uint32_t column_family_id, const rocksdb::Slice &key,

diff --git a/src/cluster/sync_migrate_context.cc b/src/cluster/sync_migrate_context.cc
@@ -68,7 +68,7 @@ void SyncMigrateContext::OnWrite(bufferevent *bev) {
   if (migrate_result_) {
     conn_->Reply(redis::SimpleString("OK"));
   } else {
-    conn_->Reply(redis::Error("ERR " + migrate_result_.Msg()));
+    conn_->Reply(redis::Error(migrate_result_));
   }
 
   timer_.reset();

diff --git a/src/commands/cmd_bloom_filter.cc b/src/commands/cmd_bloom_filter.cc
@@ -33,7 +33,7 @@ constexpr const char *errInvalidErrorRate = "error rate should be between 0 and
 constexpr const char *errInvalidCapacity = "capacity should be larger than 0";
 constexpr const char *errInvalidExpansion = "expansion should be greater or equal to 1";
 constexpr const char *errNonscalingButExpand = "nonscaling filters cannot expand";
-constexpr const char *errFilterFull = "ERR nonscaling filter is full";
+constexpr const char *errFilterFull = "nonscaling filter is full";
 }  // namespace
 
 namespace redis {
@@ -119,7 +119,7 @@ class CommandBFAdd : public Commander {
         *output = redis::Integer(0);
         break;
       case BloomFilterAddResult::kFull:
-        *output = redis::Error(errFilterFull);
+        *output = redis::Error({Status::NotOK, errFilterFull});
         break;
     }
     return Status::OK();
@@ -152,7 +152,7 @@ class CommandBFMAdd : public Commander {
           *output += redis::Integer(0);
           break;
         case BloomFilterAddResult::kFull:
-          *output += redis::Error(errFilterFull);
+          *output += redis::Error({Status::NotOK, errFilterFull});
           break;
       }
     }
@@ -248,7 +248,7 @@ class CommandBFInsert : public Commander {
           *output += redis::Integer(0);
           break;
         case BloomFilterAddResult::kFull:
-          *output += redis::Error(errFilterFull);
+          *output += redis::Error({Status::NotOK, errFilterFull});
           break;
       }
     }

diff --git a/src/commands/cmd_cluster.cc b/src/commands/cmd_cluster.cc
@@ -90,45 +90,45 @@ class CommandCluster : public Commander {
           }
         }
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "nodes") {
       std::string nodes_desc;
       Status s = srv->cluster->GetClusterNodes(&nodes_desc);
       if (s.IsOK()) {
         *output = conn->VerbatimString("txt", nodes_desc);
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "info") {
       std::string cluster_info;
       Status s = srv->cluster->GetClusterInfo(&cluster_info);
       if (s.IsOK()) {
         *output = conn->VerbatimString("txt", cluster_info);
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "import") {
       Status s = srv->cluster->ImportSlot(conn, static_cast<int>(slot_), state_);
       if (s.IsOK()) {
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "reset") {
       Status s = srv->cluster->Reset();
       if (s.IsOK()) {
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "replicas") {
       auto node_id = args_[2];
       StatusOr<std::string> s = srv->cluster->GetReplicas(node_id);
       if (s.IsOK()) {
         *output = conn->VerbatimString("txt", s.GetValue());
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else {
       return {Status::RedisExecErr, "Invalid cluster command options"};
@@ -252,23 +252,23 @@ class CommandClusterX : public Commander {
         need_persist_nodes_info = true;
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "setnodeid") {
       Status s = srv->cluster->SetNodeId(args_[2]);
       if (s.IsOK()) {
         need_persist_nodes_info = true;
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "setslot") {
       Status s = srv->cluster->SetSlotRanges(slot_ranges_, args_[4], set_version_);
       if (s.IsOK()) {
         need_persist_nodes_info = true;
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else if (subcommand_ == "version") {
       int64_t v = srv->cluster->GetVersion();
@@ -287,7 +287,7 @@ class CommandClusterX : public Commander {
         }
         *output = redis::SimpleString("OK");
       } else {
-        return {Status::RedisExecErr, s.Msg()};
+        return s;
       }
     } else {
       return {Status::RedisExecErr, "Invalid cluster command options"};

diff --git a/src/commands/cmd_hash.cc b/src/commands/cmd_hash.cc
@@ -326,16 +326,11 @@ class CommandHRangeByLex : public Commander {
         return parser.InvalidSyntax();
       }
     }
-    Status s;
     if (spec_.reversed) {
-      s = ParseRangeLexSpec(args[3], args[2], &spec_);
+      return ParseRangeLexSpec(args[3], args[2], &spec_);
     } else {
-      s = ParseRangeLexSpec(args[2], args[3], &spec_);
+      return ParseRangeLexSpec(args[2], args[3], &spec_);
     }
-    if (!s.IsOK()) {
-      return {Status::RedisParseErr, s.Msg()};
-    }
-    return Status::OK();
   }
 
   Status Execute(Server *srv, Connection *conn, std::string *output) override {

diff --git a/src/commands/cmd_key.cc b/src/commands/cmd_key.cc
@@ -480,8 +480,7 @@ class CommandSort : public Commander {
     }
 
     if (type != RedisType::kRedisList && type != RedisType::kRedisSet && type != RedisType::kRedisZSet) {
-      *output = Error("WRONGTYPE Operation against a key holding the wrong kind of value");
-      return Status::OK();
+      return {Status::RedisWrongType, "Operation against a key holding the wrong kind of value"};
     }
 
     /* When sorting a set with no sort specified, we must sort the output
@@ -508,15 +507,12 @@ class CommandSort : public Commander {
 
     switch (res) {
       case Database::SortResult::UNKNOWN_TYPE:
-        *output = redis::Error("Unknown Type");
-        break;
+        return {Status::RedisErrorNoPrefix, "Unknown Type"};
       case Database::SortResult::DOUBLE_CONVERT_ERROR:
-        *output = redis::Error("One or more scores can't be converted into double");
-        break;
+        return {Status::RedisErrorNoPrefix, "One or more scores can't be converted into double"};
       case Database::SortResult::LIMIT_EXCEEDED:
-        *output = redis::Error("The number of elements to be sorted exceeds SORT_LENGTH_LIMIT = " +
-                               std::to_string(SORT_LENGTH_LIMIT));
-        break;
+        return {Status::RedisErrorNoPrefix,
+                "The number of elements to be sorted exceeds SORT_LENGTH_LIMIT = " + std::to_string(SORT_LENGTH_LIMIT)};
       case Database::SortResult::DONE:
         if (sort_argument_.storekey.empty()) {
           std::vector<std::string> output_vec;

diff --git a/src/commands/cmd_list.cc b/src/commands/cmd_list.cc
@@ -304,7 +304,7 @@ class CommandBPop : public BlockingCommander {
         conn_->Reply(conn_->MultiBulkString({*last_key_ptr, std::move(elem)}));
       }
     } else if (!s.IsNotFound()) {
-      conn_->Reply(redis::Error("ERR " + s.ToString()));
+      conn_->Reply(redis::Error({Status::NotOK, s.ToString()}));
     }
 
     return s;
@@ -414,7 +414,7 @@ class CommandBLMPop : public BlockingCommander {
         conn_->Reply(redis::Array({redis::BulkString(chosen_key), std::move(elems_bulk)}));
       }
     } else if (!s.IsNotFound()) {
-      conn_->Reply(redis::Error("ERR " + s.ToString()));
+      conn_->Reply(redis::Error({Status::NotOK, s.ToString()}));
     }
 
     return s;
@@ -757,7 +757,7 @@ class CommandBLMove : public BlockingCommander {
     std::string elem;
     auto s = list_db.LMove(args_[1], args_[2], src_left_, dst_left_, &elem);
     if (!s.ok() && !s.IsNotFound()) {
-      conn_->Reply(redis::Error("ERR " + s.ToString()));
+      conn_->Reply(redis::Error({Status::NotOK, s.ToString()}));
       return true;
     }
 

diff --git a/src/commands/cmd_replication.cc b/src/commands/cmd_replication.cc
@@ -102,7 +102,7 @@ class CommandPSync : public Commander {
     srv->stats.IncrPSyncOKCount();
     s = srv->AddSlave(conn, next_repl_seq_);
     if (!s.IsOK()) {
-      std::string err = redis::Error(s.Msg());
+      std::string err = redis::Error(s);
       s = util::SockSend(conn->GetFD(), err, conn->GetBufferEvent());
       if (!s.IsOK()) {
         LOG(WARNING) << "failed to send error message to the replica: " << s.Msg();
@@ -230,7 +230,7 @@ class CommandFetchMeta : public Commander {
       std::string files;
       auto s = engine::Storage::ReplDataManager::GetFullReplDataInfo(srv->storage, &files);
       if (!s.IsOK()) {
-        s = util::SockSend(repl_fd, redis::Error("can't create db checkpoint"), bev);
+        s = util::SockSend(repl_fd, redis::Error({Status::RedisErrorNoPrefix, "can't create db checkpoint"}), bev);
         if (!s.IsOK()) {
           LOG(WARNING) << "[replication] Failed to send error response: " << s.Msg();
         }

diff --git a/src/commands/cmd_script.cc b/src/commands/cmd_script.cc
@@ -31,8 +31,7 @@ class CommandEvalImpl : public Commander {
  public:
   Status Execute(Server *srv, Connection *conn, std::string *output) override {
     if (evalsha && args_[1].size() != 40) {
-      *output = redis::Error(errNoMatchingScript);
-      return Status::OK();
+      return {Status::RedisNoScript, errNoMatchingScript};
     }
 
     int64_t numkeys = GET_OR_RET(ParseInt<int64_t>(args_[2], 10));