fix: backward compatibility for replication regarding HNSW index seri… (#6664)

Author: BorysTheDev

src/server/detail/save_stages_controller.cc

@@ -116,7 +116,8 @@ GenericError RdbSnapshot::Start(SaveMode save_mode, const std::string& path,
 
   is_linux_file_ = file_type & FileType::IO_URING;
   bool align_writes = (file_type & FileType::DIRECT) != 0;
-  saver_.reset(new RdbSaver(io_sink_.get(), save_mode, align_writes, snapshot_id));
+  saver_.reset(
+      new RdbSaver(io_sink_.get(), save_mode, align_writes, snapshot_id, DflyVersion::CURRENT_VER));
 
   return saver_->SaveHeader(std::move(glob_data));
 }

src/server/dflycmd.cc

@@ -683,7 +683,8 @@ OpStatus DflyCmd::StartFullSyncInThread(FlowInfo* flow, ExecutionState* exec_st,
   // of the flows also contain them.
   SaveMode save_mode =
       shard->shard_id() == 0 ? SaveMode::SINGLE_SHARD_WITH_SUMMARY : SaveMode::SINGLE_SHARD;
-  flow->saver = std::make_unique<RdbSaver>(flow->conn->socket(), save_mode, false, "");
+  flow->saver =
+      std::make_unique<RdbSaver>(flow->conn->socket(), save_mode, false, "", flow->version);
 
   flow->cleanup = [flow, shard]() {
     // socket shutdown is needed before calling saver->Cancel(). Because

src/server/rdb_save.cc

@@ -1117,7 +1117,7 @@ class RdbSaver::Impl final : public SliceSnapshot::SnapshotDataConsumerInterface
   // We pass K=sz to say how many producers are pushing data in order to maintain
   // correct closing semantics - channel is closing when K producers marked it as closed.
   Impl(bool align_writes, unsigned producers_len, CompressionMode compression_mode,
-       SaveMode save_mode, io::Sink* sink);
+       SaveMode save_mode, io::Sink* sink, DflyVersion replica_dfly_version);
 
   ~Impl();
 
@@ -1182,12 +1182,13 @@ class RdbSaver::Impl final : public SliceSnapshot::SnapshotDataConsumerInterface
   // make snapshot size smaller and opreation faster.
   CompressionMode compression_mode_;
   SaveMode save_mode_;
+  DflyVersion replica_dfly_version_ = DflyVersion::CURRENT_VER;
 };
 
 // We pass K=sz to say how many producers are pushing data in order to maintain
 // correct closing semantics - channel is closing when K producers marked it as closed.
 RdbSaver::Impl::Impl(bool align_writes, unsigned producers_len, CompressionMode compression_mode,
-                     SaveMode sm, io::Sink* sink)
+                     SaveMode sm, io::Sink* sink, DflyVersion replica_dfly_version)
     : sink_(sink),
       shard_snapshots_(producers_len),
       meta_serializer_(CompressionMode::NONE),  // Note: I think there is not need for compression
@@ -1201,6 +1202,7 @@ RdbSaver::Impl::Impl(bool align_writes, unsigned producers_len, CompressionMode
     channel_.emplace(kChannelLen, producers_len);
   }
   save_mode_ = sm;
+  replica_dfly_version_ = replica_dfly_version;
 }
 
 void RdbSaver::Impl::CleanShardSnapshots() {
@@ -1302,8 +1304,9 @@ void RdbSaver::Impl::StartSnapshotting(bool stream_journal, ExecutionState* cntx
 
 SnapshotPtr RdbSaver::Impl::CreateSliceSnapshot(EngineShard* shard, DbSlice* db_slice,
                                                 ExecutionState* cntx) {
-  return SnapshotPtr(new SliceSnapshot(compression_mode_, db_slice, this, cntx),
-                     OwnerThreadDeleter::FromShard(shard));
+  return SnapshotPtr(
+      new SliceSnapshot(compression_mode_, db_slice, this, cntx, replica_dfly_version_),
+      OwnerThreadDeleter::FromShard(shard));
 }
 
 // called on save flow
@@ -1532,8 +1535,9 @@ SnapshotPtr& RdbSaver::Impl::GetSnapshot(EngineShard* shard) {
   return shard_snapshots_[sid];
 }
 
-RdbSaver::RdbSaver(::io::Sink* sink, SaveMode save_mode, bool align_writes, std::string snapshot_id)
-    : snapshot_id_(std::move(snapshot_id)) {
+RdbSaver::RdbSaver(::io::Sink* sink, SaveMode save_mode, bool align_writes, std::string snapshot_id,
+                   DflyVersion replica_dfly_version)
+    : replica_dfly_version_(replica_dfly_version), snapshot_id_(std::move(snapshot_id)) {
   CHECK_NOTNULL(sink);
   CompressionMode compression_mode = GetDefaultCompressionMode();
   int producer_count = 0;
@@ -1561,7 +1565,8 @@ RdbSaver::RdbSaver(::io::Sink* sink, SaveMode save_mode, bool align_writes, std:
       break;
   }
   VLOG(1) << "Rdb save using compression mode:" << uint32_t(compression_mode_);
-  impl_.reset(new Impl(align_writes, producer_count, compression_mode_, save_mode, sink));
+  impl_.reset(new Impl(align_writes, producer_count, compression_mode_, save_mode, sink,
+                       replica_dfly_version_));
   save_mode_ = save_mode;
 }
 
@@ -1649,13 +1654,20 @@ error_code RdbSaver::SaveAux(const GlobalData& glob_state) {
     if (!glob_state.search_indices.empty())
       LOG(WARNING) << "Dragonfly search index data is incompatible with the RDB format";
   } else {
-    // Search index definitions (simple "index_name cmd" restore commands)
-    for (const string& s : glob_state.search_indices)
-      RETURN_ON_ERR(impl_->SaveAuxFieldStrStr("search-index", s));
+    // Search index definitions - for non-summary shards only sent to replicas >= VER6,
+    // since older replicas only expect search-index from the summary shard.
+    bool send_search_index =
+        (save_mode_ != SaveMode::SINGLE_SHARD) || (replica_dfly_version_ >= DflyVersion::VER6);
+    if (send_search_index) {
+      for (const string& s : glob_state.search_indices)
+        RETURN_ON_ERR(impl_->SaveAuxFieldStrStr("search-index", s));
+    }
 
-    // HNSW index metadata (JSON, summary only)
-    for (const string& s : glob_state.hnsw_index_metadata)
-      RETURN_ON_ERR(impl_->SaveAuxFieldStrStr("hnsw-index-metadata", s));
+    // HNSW index metadata (JSON, summary only) - only for replicas >= VER6
+    if (replica_dfly_version_ >= DflyVersion::VER6) {
+      for (const string& s : glob_state.hnsw_index_metadata)
+        RETURN_ON_ERR(impl_->SaveAuxFieldStrStr("hnsw-index-metadata", s));
+    }
 
     // Save synonyms only in summary file
     DCHECK(save_mode_ != SaveMode::SINGLE_SHARD || glob_state.search_synonyms.empty());

src/server/rdb_save.h

@@ -20,6 +20,7 @@ extern "C" {
 #include "server/journal/serializer.h"
 #include "server/journal/types.h"
 #include "server/table.h"
+#include "server/version.h"
 
 typedef struct rax rax;
 typedef struct streamCG streamCG;
@@ -97,8 +98,9 @@ class RdbSaver {
   // (corresponds to legacy, redis compatible mode)
   // if align_writes is true - writes data in aligned chunks of 4KB to fit direct I/O requirements.
   // snapshot_id - allows to identify that group of files belongs to the same snapshot
+  // replica_dfly_version - upper bound for conditional serialization of new features.
   explicit RdbSaver(::io::Sink* sink, SaveMode save_mode, bool align_writes,
-                    std::string snapshot_id);
+                    std::string snapshot_id, DflyVersion replica_dfly_version);
 
   ~RdbSaver();
 
@@ -159,6 +161,7 @@ class RdbSaver {
   std::unique_ptr<Impl> impl_;
   SaveMode save_mode_;
   CompressionMode compression_mode_;
+  DflyVersion replica_dfly_version_ = DflyVersion::CURRENT_VER;
   std::string snapshot_id_;
 };
 

src/server/snapshot.cc

@@ -45,10 +45,12 @@ constexpr size_t kMinBlobSize = 8_KB;
 }  // namespace
 
 SliceSnapshot::SliceSnapshot(CompressionMode compression_mode, DbSlice* slice,
-                             SnapshotDataConsumerInterface* consumer, ExecutionState* cntx)
+                             SnapshotDataConsumerInterface* consumer, ExecutionState* cntx,
+                             DflyVersion replica_dfly_version)
     : db_slice_(slice),
       db_array_(slice->databases()),
       compression_mode_(compression_mode),
+      replica_dfly_version_(replica_dfly_version),
       consumer_(consumer),
       cntx_(cntx) {
   tl_slice_snapshots.insert(this);
@@ -189,7 +191,8 @@ void SliceSnapshot::SerializeIndexMapping(
 
 void SliceSnapshot::SerializeIndexMappings() {
 #ifdef WITH_SEARCH
-  if (SaveMode() == dfly::SaveMode::RDB || !absl::GetFlag(FLAGS_serialize_hnsw_index)) {
+  if (SaveMode() == dfly::SaveMode::RDB || !absl::GetFlag(FLAGS_serialize_hnsw_index) ||
+      replica_dfly_version_ < DflyVersion::VER6) {
     return;
   }
 
@@ -220,7 +223,7 @@ void SliceSnapshot::SerializeGlobalHnswIndices() {
 #ifdef WITH_SEARCH
   // Serialize HNSW global indices for shard 0 only
   if (db_slice_->shard_owner()->shard_id() != 0 || SaveMode() == dfly::SaveMode::RDB ||
-      !absl::GetFlag(FLAGS_serialize_hnsw_index)) {
+      !absl::GetFlag(FLAGS_serialize_hnsw_index) || replica_dfly_version_ < DflyVersion::VER6) {
     return;
   }
 

src/server/snapshot.h

@@ -65,7 +65,8 @@ class SliceSnapshot : public journal::JournalConsumerInterface {
   };
 
   SliceSnapshot(CompressionMode compression_mode, DbSlice* slice,
-                SnapshotDataConsumerInterface* consumer, ExecutionState* cntx);
+                SnapshotDataConsumerInterface* consumer, ExecutionState* cntx,
+                DflyVersion replica_dfly_version);
   ~SliceSnapshot();
 
   static size_t GetThreadLocalMemoryUsage();
@@ -177,6 +178,7 @@ class SliceSnapshot : public journal::JournalConsumerInterface {
 
   bool use_background_mode_ = false;
   bool use_snapshot_version_ = true;
+  DflyVersion replica_dfly_version_ = DflyVersion::CURRENT_VER;
 
   uint64_t rec_id_ = 1, last_pushed_id_ = 0;
 

src/server/version.h

@@ -36,8 +36,14 @@ enum class DflyVersion {
   // - Support partial sync from different master
   VER5,
 
+  // 1.37 <= ver
+  // - Per-shard search index definitions (search-index AUX on every flow)
+  // - HNSW index serialization opcodes (RDB_OPCODE_VECTOR_INDEX, RDB_OPCODE_SHARD_DOC_INDEX)
+  // - hnsw-index-metadata AUX field
+  VER6,
+
   // Always points to the latest version
-  CURRENT_VER = VER5,
+  CURRENT_VER = VER6,
 };
 
 }  // namespace dfly

tests/dragonfly/replication_test.py

@@ -2,6 +2,7 @@
 import platform
 import shutil
 import signal
+import struct
 import tarfile
 import time
 import urllib.request
@@ -2623,6 +2624,108 @@ async def check_if_empty():
     assert await c_replica.execute_command(f"dbsize") == 0
 
 
+async def test_replicate_search_index_to_old_replica(df_factory: DflyInstanceFactory):
+    """
+    Test that a new master with search indices (including HNSW vector index) can
+    replicate to a v1.35 replica. This verifies backward compatibility of replication
+    when search indices are defined, ensuring the replica receives the data without
+    errors from new RDB AUX fields (search-index, hnsw-index-metadata, HNSW opcodes).
+    """
+    cpu = platform.processor()
+    if cpu != "x86_64":
+        pytest.skip(f"Supported only on x64, running on {cpu}")
+
+    dfly_version = "v1.35.1"
+    released_dfly_path = download_dragonfly_release(dfly_version)
+
+    # New master (current version) with search index
+    master = df_factory.create(proactor_threads=2)
+    # Old replica (v1.35)
+    replica = df_factory.create(
+        version=1.35,
+        path=released_dfly_path,
+        proactor_threads=2,
+    )
+
+    df_factory.start_all([master, replica])
+
+    c_master = master.client()
+    c_replica = replica.client()
+
+    # Create a search index with HNSW vector field on the new master
+    await c_master.execute_command(
+        "FT.CREATE",
+        "test_idx",
+        "ON",
+        "HASH",
+        "PREFIX",
+        "1",
+        "item:",
+        "SCHEMA",
+        "name",
+        "TEXT",
+        "price",
+        "NUMERIC",
+        "SORTABLE",
+        "category",
+        "TAG",
+        "embedding",
+        "VECTOR",
+        "HNSW",
+        "6",
+        "TYPE",
+        "FLOAT32",
+        "DIM",
+        "2",
+        "DISTANCE_METRIC",
+        "L2",
+    )
+
+    # Insert test data with vector embeddings
+    for i in range(100):
+        category = "electronics" if i % 2 == 0 else "clothing"
+        embedding = struct.pack("<2f", float(i), float(i * 2))
+        await c_master.hset(
+            f"item:{i}",
+            mapping={
+                "name": f"Product {i}",
+                "price": str(i * 10),
+                "category": category,
+                "embedding": embedding,
+            },
+        )
+
+    # Verify data and index on master
+    assert await c_master.dbsize() == 100
+    master_idx = c_master.ft("test_idx")
+    text_result = await master_idx.search("Product 50")
+    assert text_result.total >= 1
+
+    # Verify KNN search on master
+    query_vec = struct.pack("<2f", 50.0, 100.0)
+    knn_result = await c_master.execute_command(
+        "FT.SEARCH", "test_idx", "*=>[KNN 2 @embedding $vec]", "PARAMS", "2", "vec", query_vec
+    )
+    assert knn_result[0] >= 1
+    assert "item:50" in knn_result
+
+    # Start replication from new master to old replica
+    await c_replica.execute_command(f"REPLICAOF localhost {master.port}")
+    await wait_available_async(c_replica)
+
+    # Verify data replicated successfully
+    assert await c_replica.dbsize() == 100
+    assert await c_replica.hget("item:0", "name") == "Product 0"
+    assert await c_replica.hget("item:99", "name") == "Product 99"
+
+    # Verify KNN search works on old replica (index rebuilt from replicated data)
+    knn_result = await c_replica.execute_command(
+        "FT.SEARCH", "test_idx", "*=>[KNN 2 @embedding $vec]", "PARAMS", "2", "vec", query_vec
+    )
+    assert knn_result[0] >= 1
+    assert "item:50" in knn_result
+
+
 async def test_replicating_mc_flags(df_factory):
     master = df_factory.create(memcached_port=11211, proactor_threads=1)
     replica = df_factory.create(