fix(hgraph): backport duplicate handling to v0.18

docs/hgraph.md

@@ -225,10 +225,21 @@ means that the index uses PQ quantization with 64 subspaces, enables reordering
 - **Optional Values**: 1 to INT_MAX
 - **Default Value**: Must be provided (no default value)
 
+### max_duplicates_per_group
+- **Parameter Type**: int
+- **Parameter Description**: Maximum number of additional duplicate IDs returned for each
+  duplicate group during KNN graph search
+- **Optional Values**: -1 (unlimited), 0 (do not expand duplicate IDs), or any positive integer
+- **Default Value**: -1
+- **Notes**: Only applies when `support_duplicate` is enabled; range search keeps its existing
+  expansion behavior
+
 ## Examples for Search Parameter String
 ```json
 "hgraph": {
-    "ef_search": 200
+    "ef_search": 200,
+    "max_duplicates_per_group": 2
 }
 ```
-means that the search will use an ef_search value of 200 to control the search quality and performance trade-off.
+means that the search uses an ef_search value of 200 and returns at most two additional IDs from
+each duplicate group.

include/vsag/constants.h

@@ -184,6 +184,7 @@ extern const char* const HGRAPH_PRECISE_IO_TYPE;
 extern const char* const HGRAPH_PRECISE_FILE_PATH;
 extern const char* const HGRAPH_PARAMETER_EF_RUNTIME;
 extern const char* const HGRAPH_PARAMETER_HOPS_LIMIT;
+extern const char* const HGRAPH_PARAMETER_MAX_DUPLICATES_PER_GROUP;
 extern const char* const HGRAPH_EXTRA_INFO_SIZE;
 extern const char* const HGRAPH_SUPPORT_DUPLICATE;
 extern const char* const HGRAPH_DUPLICATE_DISTANCE_THRESHOLD;

src/algorithm/hgraph.cpp

@@ -933,10 +933,20 @@ HGraph::KnnSearch(const DatasetPtr& query,
     auto search_result = DistanceHeap::MakeInstanceBySize<true, false>(ctx.alloc, k);
     const auto* query_data = get_data(query);
     if (is_last_filter) {
+        if (const auto* pending_duplicates = iter_filter_ctx->GetPendingDuplicates();
+            pending_duplicates != nullptr) {
+            for (const auto& [pending_id, pending_dist] : *pending_duplicates) {
+                if (iter_filter_ctx->CheckPoint(pending_id)) {
+                    search_result->Push(pending_dist, pending_id);
+                }
+            }
+        }
         while (!iter_filter_ctx->Empty()) {
             uint32_t cur_inner_id = iter_filter_ctx->GetTopID();
             float cur_dist = iter_filter_ctx->GetTopDist();
-            search_result->Push(cur_dist, cur_inner_id);
+            if (not iter_filter_ctx->IsPendingDuplicate(cur_inner_id)) {
+                search_result->Push(cur_dist, cur_inner_id);
+            }
             iter_filter_ctx->PopDiscard();
         }
     } else {
@@ -964,6 +974,7 @@ HGraph::KnnSearch(const DatasetPtr& query,
         search_param.is_inner_id_allowed = ft;
         search_param.topk = static_cast<int64_t>(search_param.ef);
         search_param.consider_duplicate = this->label_table_->CompressDuplicateData();
+        search_param.max_duplicates_per_group = params.max_duplicates_per_group;
         search_param.parallel_search_thread_count = params.parallel_search_thread_count;
         search_param.min_distance = params.min_distance;
 
@@ -1410,6 +1421,9 @@ HGraph::Serialize(StreamWriter& writer) const {
         if (this->use_attribute_filter_ and this->attr_filter_index_ != nullptr) {
             this->attr_filter_index_->Serialize(writer);
         }
+        if (this->label_table_->CompressDuplicateData()) {
+            this->label_table_->SerializeDuplicateRecords(writer);
+        }
         return;
     }
 
@@ -1442,10 +1456,53 @@ HGraph::Serialize(StreamWriter& writer) const {
     footer->Write(writer);
 }
 
+void
+HGraph::Deserialize(std::istream& in_stream) {
+    if (not this->use_old_serial_format_) {
+        InnerIndexInterface::Deserialize(in_stream);
+        return;
+    }
+
+    try {
+        uint64_t cursor = 0;
+        auto read_func = [&](uint64_t offset, uint64_t size, void* data) {
+            if (offset != cursor) {
+                throw VsagException(ErrorType::UNSUPPORTED_INDEX_OPERATION,
+                                    "v0.14 sequential stream does not support seek");
+            }
+            in_stream.read(static_cast<char*>(data), static_cast<int64_t>(size));
+            if (in_stream.gcount() != static_cast<int64_t>(size)) {
+                throw VsagException(
+                    ErrorType::READ_ERROR,
+                    fmt::format("Attempted to read: {} bytes. Remaining content size: {} bytes.",
+                                size,
+                                in_stream.gcount()));
+            }
+            cursor += size;
+        };
+        ReadFuncStreamReader reader(read_func, 0, 0);
+        this->deserialize(reader, true);
+        if (reader.GetCursor() != cursor) {
+            throw VsagException(ErrorType::UNSUPPORTED_INDEX_OPERATION,
+                                "v0.14 sequential stream does not support seek");
+        }
+    } catch (const std::bad_alloc& e) {
+        throw VsagException(ErrorType::NO_ENOUGH_MEMORY, "failed to Deserialize: ", e.what());
+    }
+}
+
 void
 HGraph::Deserialize(StreamReader& reader) {
-    // try to deserialize footer (only in new version)
-    auto footer = Footer::Parse(reader);
+    this->deserialize(reader, false);
+}
+
+void
+HGraph::deserialize(StreamReader& reader, bool force_v0_14) {
+    FooterPtr footer = nullptr;
+    if (not force_v0_14) {
+        // try to deserialize footer (only in new version)
+        footer = Footer::Parse(reader);
+    }
 
     if (footer == nullptr) {  // old format, DON'T EDIT, remove in the future
         logger::debug("parse with v0.14 version format");
@@ -1474,6 +1531,11 @@ HGraph::Deserialize(StreamReader& reader) {
         if (this->use_attribute_filter_ and this->attr_filter_index_ != nullptr) {
             this->attr_filter_index_->Deserialize(reader);
         }
+        if (this->label_table_->CompressDuplicateData()) {
+            const auto logical_element_count =
+                static_cast<uint64_t>(this->label_table_->GetTotalCount());
+            this->label_table_->DeserializeDuplicateRecords(reader, logical_element_count);
+        }
     } else {  // create like `else if ( ver in [v0.15, v0.17] )` here if need in the future
         logger::debug("parse with new version format");
 
@@ -2222,6 +2284,7 @@ HGraph::SearchWithRequest(const SearchRequest& request) const {
             search_param.topk, static_cast<int64_t>(static_cast<float>(k) * params.topk_factor));
     }
     search_param.consider_duplicate = true;
+    search_param.max_duplicates_per_group = params.max_duplicates_per_group;
     if (params.enable_time_record) {
         search_param.time_cost = std::make_shared<Timer>();
         search_param.time_cost->SetThreshold(params.timeout_ms);

src/algorithm/hgraph.h

@@ -15,6 +15,7 @@
 
 #pragma once
 
+#include <istream>
 #include <random>
 #include <shared_mutex>
 #include <string>
@@ -79,6 +80,9 @@ class HGraph : public InnerIndexInterface {
     DatasetPtr
     CalDistanceById(const float* query, const int64_t* ids, int64_t count) const override;
 
+    void
+    Deserialize(std::istream& in_stream) override;
+
     void
     Deserialize(StreamReader& reader) override;
 
@@ -285,6 +289,9 @@ class HGraph : public InnerIndexInterface {
     void
     deserialize_label_info(StreamReader& reader) const;
 
+    void
+    deserialize(StreamReader& reader, bool force_v0_14);
+
     // used in version [0.12.*, 0.14.*]
     void
     serialize_basic_info_v0_14(StreamWriter& writer) const;

src/algorithm/hgraph_parameter.cpp

@@ -15,6 +15,8 @@
 
 #include "hgraph_parameter.h"
 
+#include <nlohmann/json.hpp>
+
 #include "datacell/extra_info_datacell_parameter.h"
 #include "datacell/flatten_datacell_parameter.h"
 #include "datacell/graph_datacell_parameter.h"
@@ -223,6 +225,22 @@ HGraphSearchParameters::FromJson(const std::string& json_string) {
     if (params[INDEX_TYPE_HGRAPH].Contains("min_distance")) {
         obj.min_distance = params[INDEX_TYPE_HGRAPH]["min_distance"].GetFloat();
     }
+    if (params[INDEX_TYPE_HGRAPH].Contains(HGRAPH_PARAMETER_MAX_DUPLICATES_PER_GROUP)) {
+        const auto& max_duplicates =
+            params[INDEX_TYPE_HGRAPH][HGRAPH_PARAMETER_MAX_DUPLICATES_PER_GROUP];
+        CHECK_ARGUMENT(max_duplicates.IsNumberInteger(),
+                       "max_duplicates_per_group must be an integer");
+        const auto* max_duplicates_json = max_duplicates.GetInnerJson();
+        if (max_duplicates_json->is_number_unsigned()) {
+            CHECK_ARGUMENT(max_duplicates_json->get<uint64_t>() <=
+                               static_cast<uint64_t>(std::numeric_limits<int64_t>::max()),
+                           "max_duplicates_per_group exceeds int64_t range");
+        }
+        obj.max_duplicates_per_group = max_duplicates.GetInt();
+        CHECK_ARGUMENT(obj.max_duplicates_per_group >= -1,
+                       fmt::format("max_duplicates_per_group({}) must be >= -1",
+                                   obj.max_duplicates_per_group));
+    }
 
     return obj;
 }

src/algorithm/hgraph_parameter.h

@@ -82,6 +82,9 @@ class HGraphSearchParameters : public IndexSearchParameter {
     bool use_reorder{false};
     bool use_extra_info_filter{false};
     float min_distance{std::numeric_limits<float>::lowest()};
+    // Maximum additional duplicate IDs returned per group during KNN graph search.
+    // -1 means unlimited; 0 disables duplicate expansion.
+    int64_t max_duplicates_per_group{-1};
 
 private:
     HGraphSearchParameters() = default;

src/algorithm/hgraph_parameter_test.cpp

@@ -204,6 +204,24 @@ TEST_CASE("HGraph maps support_duplicate to graph parameter", "[ut][HGraphParame
     REQUIRE(typed_param->duplicate_distance_threshold == 0.25F);
 }
 
+TEST_CASE("HGraph Search Parameters validate max_duplicates_per_group",
+          "[ut][HGraphSearchParameters][duplicate]") {
+    REQUIRE(vsag::HGraphSearchParameters::FromJson(R"({"hgraph": {"ef_search": 32}})")
+                .max_duplicates_per_group == -1);
+    REQUIRE(vsag::HGraphSearchParameters::FromJson(
+                R"({"hgraph": {"ef_search": 32, "max_duplicates_per_group": 0}})")
+                .max_duplicates_per_group == 0);
+    REQUIRE(vsag::HGraphSearchParameters::FromJson(
+                R"({"hgraph": {"ef_search": 32, "max_duplicates_per_group": 2}})")
+                .max_duplicates_per_group == 2);
+    REQUIRE_THROWS(vsag::HGraphSearchParameters::FromJson(
+        R"({"hgraph": {"ef_search": 32, "max_duplicates_per_group": -2}})"));
+    REQUIRE_THROWS(vsag::HGraphSearchParameters::FromJson(
+        R"({"hgraph": {"ef_search": 32, "max_duplicates_per_group": 1.5}})"));
+    REQUIRE_THROWS(vsag::HGraphSearchParameters::FromJson(
+        R"({"hgraph": {"ef_search": 32, "max_duplicates_per_group": 9223372036854775808}})"));
+}
+
 TEST_CASE("HGraph maps label_remap_type to inner index parameter", "[ut][HGraphParameter]") {
     auto param = vsag::JsonType::Parse(R"({
         "base_quantization_type": "fp32",

src/analyzer/hgraph_analyzer.cpp

@@ -508,6 +508,9 @@ HGraphAnalyzer::GetDegreeDistribution() {
     Vector<uint32_t> in_degree(this->total_count_, allocator_);
     Vector<uint32_t> out_degree(this->total_count_, allocator_);
     for (InnerIdType i = 0; i < this->total_count_; ++i) {
+        if (is_duplicate_ids_[i]) {
+            continue;
+        }
         Vector<InnerIdType> neighbors(allocator_);
         hgraph_->bottom_graph_->GetNeighbors(i, neighbors);
         out_degree[i] = neighbors.size();

src/constants.cpp

@@ -164,6 +164,7 @@ const char* const HGRAPH_PRECISE_IO_TYPE = "precise_io_type";
 const char* const HGRAPH_PRECISE_FILE_PATH = "precise_file_path";
 const char* const HGRAPH_PARAMETER_EF_RUNTIME = "ef_search";
 const char* const HGRAPH_PARAMETER_HOPS_LIMIT = "hops_limit";
+const char* const HGRAPH_PARAMETER_MAX_DUPLICATES_PER_GROUP = "max_duplicates_per_group";
 const char* const HGRAPH_EXTRA_INFO_SIZE = "extra_info_size";
 const char* const HGRAPH_SUPPORT_DUPLICATE = "support_duplicate";
 const char* const HGRAPH_DUPLICATE_DISTANCE_THRESHOLD = "duplicate_distance_threshold";

src/impl/inner_search_param.h

@@ -64,6 +64,7 @@ class InnerSearchParam {
 
     // use in search process with duplicate ids
     bool consider_duplicate{false};
+    int64_t max_duplicates_per_group{-1};
 
     // skip results with dist <= min_distance (for search iterator)
     float min_distance{std::numeric_limits<float>::lowest()};
@@ -87,6 +88,8 @@ class InnerSearchParam {
             factor = other.factor;
             first_order_scan_ratio = other.first_order_scan_ratio;
             parallel_search_thread_count = other.parallel_search_thread_count;
+            consider_duplicate = other.consider_duplicate;
+            max_duplicates_per_group = other.max_duplicates_per_group;
         }
         return *this;
     }