feat: add CSS option for FT.CREATE (#6057)

* feat: add CSS option for FT.CREATE

Author: BorysTheDev

src/server/cluster/cluster_config.h

@@ -32,6 +32,10 @@ class ClusterConfig {
   bool IsMySlot(SlotId id) const;
   bool IsMySlot(std::string_view key) const;
 
+  const std::string& MyId() const {
+    return my_id_;
+  }
+
   // Returns the master configured for `id`.
   ClusterNodeInfo GetMasterNodeForSlot(SlotId id) const;
 

src/server/cluster/coordinator.cc

@@ -0,0 +1,86 @@
+// Copyright 2025, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#include "server/cluster/coordinator.h"
+
+#include "base/logging.h"
+#include "facade/redis_parser.h"
+#include "facade/socket_utils.h"
+#include "server/cluster/cluster_config.h"
+
+using namespace std;
+using namespace facade;
+
+namespace dfly::cluster {
+
+class Coordinator::CrossShardClient : private ProtocolClient {
+ public:
+  CrossShardClient(std::string host, uint16_t port) : ProtocolClient(std::move(host), port) {
+  }
+
+  using ProtocolClient::CloseSocket;
+  ~CrossShardClient() {
+    CloseSocket();
+  }
+
+  void Init() {
+    VLOG(1) << "Resolving host DNS to " << server().Description();
+    if (error_code ec = ResolveHostDns(); ec) {
+      LOG(WARNING) << "Could not resolve host DNS to " << server().Description() << ": "
+                   << ec.message();
+      exec_st_.ReportError(GenericError(ec, "Could not resolve host dns."));
+      return;
+    }
+    VLOG(1) << "Start coordinator connection to " << server().Description();
+    auto timeout = 3000ms;  // TODO add flag;
+    if (auto ec = ConnectAndAuth(timeout, &exec_st_); ec) {
+      LOG(WARNING) << "Couldn't connect to " << server().Description() << ": " << ec.message()
+                   << ", socket state: " << GetSocketInfo(Sock()->native_handle());
+      exec_st_.ReportError(GenericError(ec, "Couldn't connect to source."));
+      return;
+    }
+
+    ResetParser(RedisParser::Mode::CLIENT);
+  }
+
+  void Cancel() {
+    ShutdownSocket();
+  }
+
+  void SendCommand(std::string_view cmd) {
+    if (auto ec = SendCommandAndReadResponse(cmd); ec) {
+      LOG(WARNING) << "Coordinator could not send command to : " << server().Description() << ": "
+                   << ec.message() << ", socket state: " << GetSocketInfo(Sock()->native_handle());
+      exec_st_.ReportError(GenericError(ec, "Could not send command."));
+    }
+    // add response processing
+  }
+};
+
+Coordinator& Coordinator::Current() {
+  static Coordinator instance;
+  return instance;
+}
+
+void Coordinator::DispatchAll(std::string_view command) {
+  auto cluster_config = ClusterConfig::Current();
+  if (!cluster_config) {
+    VLOG(2) << "No cluster config found for coordinator plan creation.";
+    return;
+  }
+  VLOG(2) << "Dispatching command to all shards: " << command;
+  auto shards_config = cluster_config->GetConfig();
+
+  std::vector<std::unique_ptr<CrossShardClient>> clients;
+  for (const auto& shard : shards_config) {
+    if (shard.master.id == cluster_config->MyId()) {
+      continue;
+    }
+    clients.emplace_back(std::make_unique<CrossShardClient>(shard.master.ip, shard.master.port));
+    clients.back()->Init();
+    clients.back()->SendCommand(std::string(command));
+  }
+}
+
+}  // namespace dfly::cluster

src/server/cluster/coordinator.h

@@ -0,0 +1,24 @@
+// Copyright 2025, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+#pragma once
+
+#include "server/cluster/cluster_defs.h"
+#include "server/protocol_client.h"
+
+namespace dfly::cluster {
+
+// Coordinator needs to create and manage connections between nodes in the cluster for cross shard
+// commands. All cross-shard commands are dispatched through the Coordinator.
+// It can be used to exeute commands on all shards or specific shards.
+class Coordinator {
+ public:
+  static Coordinator& Current();
+  void DispatchAll(std::string_view command);
+
+ private:
+  Coordinator() = default;
+  class CrossShardClient;
+};
+
+}  // namespace dfly::cluster

src/server/search/CMakeLists.txt

@@ -4,7 +4,8 @@ if (NOT WITH_SEARCH)
   return()
 endif()
 
-add_library(dfly_search_server aggregator.cc doc_accessors.cc doc_index.cc search_family.cc index_join.cc)
+add_library(dfly_search_server aggregator.cc doc_accessors.cc doc_index.cc search_family.cc index_join.cc
+  ../cluster/coordinator.cc)
 target_link_libraries(dfly_search_server dfly_transaction dragonfly_lib dfly_facade redis_lib jsonpath TRDP::jsoncons)
 
 

src/server/search/search_family.cc

@@ -9,6 +9,7 @@
 #include <absl/strings/ascii.h>
 #include <absl/strings/match.h>
 #include <absl/strings/str_format.h>
+#include <absl/strings/str_join.h>
 #include <absl/strings/str_split.h>
 #include <absl/strings/string_view.h>
 
@@ -24,6 +25,8 @@
 #include "facade/error.h"
 #include "facade/reply_builder.h"
 #include "server/acl/acl_commands_def.h"
+#include "server/cluster/cluster_config.h"
+#include "server/cluster/coordinator.h"
 #include "server/command_registry.h"
 #include "server/config_registry.h"
 #include "server/conn_context.h"
@@ -316,7 +319,7 @@ ParseResult<bool> ParseSchema(CmdArgParser* parser, DocIndex* index) {
 #pragma GCC diagnostic pop
 #endif
 
-ParseResult<DocIndex> ParseCreateParams(CmdArgParser* parser) {
+ParseResult<DocIndex> CreateDocIndex(CmdArgParser* parser) {
   DocIndex index{};
 
   while (parser->HasNext()) {
@@ -1047,7 +1050,9 @@ void SearchFamily::FtCreate(CmdArgList args, const CommandContext& cmd_cntx) {
   CmdArgParser parser{args};
   string_view idx_name = parser.Next();
 
-  auto parsed_index = ParseCreateParams(&parser);
+  bool is_cross_shard = parser.Check("CSS");
+
+  auto parsed_index = CreateDocIndex(&parser);
   if (SendErrorIfOccurred(parsed_index, &parser, builder)) {
     return;
   }
@@ -1069,6 +1074,15 @@ void SearchFamily::FtCreate(CmdArgList args, const CommandContext& cmd_cntx) {
     return builder->SendError("Index already exists");
   }
 
+  if (!is_cross_shard && IsClusterEnabled()) {
+    std::string args_str = absl::StrJoin(args.subspan(1), " ");
+    std::string cmd = absl::StrCat("FT.CREATE ", idx_name, " CSS ", args_str);
+
+    // TODO add processing of the reply to make sure index was created successfully on all shards,
+    // and prevent simultaneous creation of the same index.
+    cluster::Coordinator::Current().DispatchAll(cmd);
+  }
+
   auto idx_ptr = make_shared<DocIndex>(std::move(parsed_index).value());
   cmd_cntx.tx->Execute(
       [idx_name, idx_ptr](auto* tx, auto* es) {
@@ -1283,6 +1297,8 @@ void SearchFamily::FtSearch(CmdArgList args, const CommandContext& cmd_cntx) {
   string_view index_name = parser.Next();
   string_view query_str = parser.Next();
 
+  bool is_cross_shard = parser.Check("CSS");
+
   auto* builder = cmd_cntx.rb;
   auto params = ParseSearchParams(&parser);
   if (SendErrorIfOccurred(params, &parser, builder))
@@ -1295,6 +1311,13 @@ void SearchFamily::FtSearch(CmdArgList args, const CommandContext& cmd_cntx) {
         absl::StrCat("Query string is too long, max length is ", max_query_bytes, " bytes"));
   }
 
+  if (!is_cross_shard && IsClusterEnabled()) {
+    std::string args_str = absl::StrJoin(args.subspan(2), " ");
+    std::string cmd = absl::StrCat("FT.SEARCH ", index_name, " ", query_str, " CSS ", args_str);
+
+    cluster::Coordinator::Current().DispatchAll(cmd);
+  }
+
   search::SearchAlgorithm search_algo;
   if (!search_algo.Init(query_str, &params->query_params, &params->optional_filters))
     return builder->SendError("Query syntax error");

tests/dragonfly/cluster_test.py

@@ -181,6 +181,14 @@ async def wait_for_status(admin_client, node_id, status, timeout=10):
             assert len(states) != 0 and all(state[2] in status for state in states), states
 
 
+async def wait_for_ft_index_creation(client, idx_name, timeout=5):
+    get_status = lambda: client.execute_command("FT.INFO", idx_name)
+
+    async for states, breaker in tick_timer(get_status, timeout=timeout):
+        with breaker:
+            assert len(states) != 0, states
+
+
 async def wait_for_error(admin_client, node_id, error, timeout=10):
     get_status = lambda: admin_client.execute_command(
         "DFLYCLUSTER", "SLOT-MIGRATION-STATUS", node_id
@@ -3442,3 +3450,36 @@ async def test_replica_takeover_moved(
     await m1.client.execute_command(f"replicaof localhost {r1.instance.port}")
     await check_all_replicas_finished([m1.client], r1.client)
     assert await m1.client.execute_command("GET newk") == "foo"
+
+
+@dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
+async def test_SearchRequestDistribution(df_factory: DflyInstanceFactory):
+    """
+    Create cluster of 3 nodes.
+    Send FT.CREATE to first node and check that index was created on all nodes.
+    """
+
+    instances = [
+        df_factory.create(port=next(next_port), admin_port=next(next_port), vmodule="coordinator=2")
+        for i in range(3)
+    ]
+
+    df_factory.start_all(instances)
+
+    nodes = [(await create_node_info(instance)) for instance in instances]
+    nodes[0].slots = [(0, 5259)]
+    nodes[1].slots = [(5260, 10519)]
+    nodes[2].slots = [(10520, 16383)]
+
+    await push_config(json.dumps(generate_config(nodes)), [node.admin_client for node in nodes])
+
+    assert (
+        await nodes[0].client.execute_command(
+            "FT.CREATE", "idx", "ON", "HASH", "SCHEMA", "title", "TEXT"
+        )
+        == "OK"
+    )
+
+    await asyncio.sleep(3)
+    for node in nodes:
+        await wait_for_ft_index_creation(node.client, "idx")