perf(hgraph): share vector storage for PiPNN duplicates

docs/docs/en/src/indexes/hgraph.md

@@ -90,7 +90,7 @@ most users need; the exhaustive list is in [Index Parameters](../resources/index
 | `rabitq_fused_datacell` | bool | `false` | Fuse the bottom HGraph node and RaBitQ split codes into one in-memory record. Requires L2/IP, flat in-memory graph storage, RaBitQ x+y split codes with x in `[1, 4]`, and the other constraints described in [RaBitQ x+y split](../quantization/rabitq_split.md). |
 | `train_sample_count` | int | `65536` | Maximum number of vectors sampled for quantizer training; must be at least `512` when set explicitly. |
 | `build_thread_count` | int | `100` | Threads used to parallelise build |
-| `support_duplicate` | bool | `false` | Enable duplicate-ID detection on insert |
+| `support_duplicate` | bool | `false` | Group duplicate vectors while retaining their individual labels |
 | `deduplicate_storage` | bool | `false` | Share vector storage between duplicates; requires `support_duplicate: true` |
 | `duplicate_distance_threshold` | float | `0.0` | Duplicate-detection distance threshold. When greater than `0`, deduplicate by the nearest candidate distance; when `0`, compare codes with the nearest candidate first, then check the remaining returned build candidates until a match is found |
 | `support_remove` | bool | `false` | Enable graph delete-tracking metadata used by mark-remove recovery paths |
@@ -122,8 +122,8 @@ full `Build`. The PiPNN builder accepts dense `float32` input with `metric_type`
 `"ip"`, or `"cosine"`. It builds the bottom graph from the original build vectors, then reuses
 HGraph's route layers, storage, search, filtering, reordering, incremental `Add`, removal, and
 serialization paths. The persistent base storage may use a supported quantizer such as `sq8`,
-including RaBitQ with SQ8 reorder. Cache-assisted build and deduplicated vector storage are not
-supported with PiPNN. The `pipnn_*` parameters in the table above tune this builder;
+including RaBitQ with SQ8 reorder. Cache-assisted build is not supported with PiPNN.
+The `pipnn_*` parameters in the table above tune this builder;
 `ef_construction` does not. The existing `alpha` parameter controls PiPNN's final robust pruning.
 
 `build_thread_count` parallelizes vector preparation, partitioning, candidate generation, and
@@ -146,8 +146,14 @@ The policy and step are serialized. Loading an enabled index requires matching p
 
 Set both `support_duplicate: true` and `deduplicate_storage: true` to let duplicate
 vectors share one physical code slot while retaining their individual labels. This option
-currently supports only dense-vector HGraph indexes using `graph_type: "nsw"`; it is not
-available for `graph_type: "odescent"` or `graph_type: "pipnn"`.
+supports dense-vector HGraph indexes using `graph_type: "nsw"` or `"pipnn"`;
+`graph_type: "odescent"` is not supported.
+
+For PiPNN, initial `Build` groups byte-identical FP32 input rows and encodes each
+representative once. All labels, attributes, and extra information remain independent.
+This also works with adaptive alpha. `duplicate_distance_threshold` does not affect
+PiPNN batch grouping; later incremental `Add` retains its existing threshold/code-based
+duplicate detection. Quantizer training still uses the original training sample.
 
 The following operations and configurations are not supported while storage deduplication
 is enabled:

docs/docs/zh/src/indexes/hgraph.md

@@ -84,7 +84,7 @@ auto result = index->KnnSearch(
 | `rabitq_fused_datacell` | bool | `false` | 将底层 HGraph 节点与 RaBitQ split 编码融合到一条内存记录中；要求 L2/IP、flat 内存图、x 在 `[1, 4]` 的 RaBitQ x+y split 编码，并满足 [RaBitQ x+y split](../quantization/rabitq_split.md) 中的其他约束 |
 | `train_sample_count` | int | `65536` | 量化器训练的最大采样向量数；显式配置时最小为 `512` |
 | `build_thread_count` | int | `100` | 构建阶段并发线程数 |
-| `support_duplicate` | bool | `false` | 是否在插入时做重复 ID 检测 |
+| `support_duplicate` | bool | `false` | 合并重复向量的图节点，并保留各自的标签 |
 | `deduplicate_storage` | bool | `false` | 让重复向量共享存储；需同时设置 `support_duplicate: true` |
 | `duplicate_distance_threshold` | float | `0.0` | 重复判定距离阈值。大于 `0` 时按最近候选的距离判重；等于 `0` 时先比较最近候选的编码，不匹配则继续检查构建搜索返回的其余候选，直到找到匹配 |
 | `support_remove` | bool | `false` | 是否启用 mark-remove 恢复路径所需的图删除追踪元数据 |
@@ -113,8 +113,8 @@ auto result = index->KnnSearch(
 [PiPNN](https://arxiv.org/abs/2602.21247)。PiPNN 构建器接收 `dtype: "float32"`、
 `metric_type: "l2"`、`"ip"` 或 `"cosine"` 的稠密向量输入，从原始构建向量生成底层图，并复用
 HGraph 现有的路由层、向量存储、搜索、过滤、精排、增量 `Add`、删除和序列化路径。持久化底层
-存储可以使用 `sq8` 等受支持的量化器，包括 RaBitQ 配合 SQ8 精排。缓存辅助构建和向量存储去重
-暂不支持 PiPNN。上表的 `pipnn_*` 参数用于调节该构建器，`ef_construction` 不生效；
+存储可以使用 `sq8` 等受支持的量化器，包括 RaBitQ 配合 SQ8 精排。缓存辅助构建暂不支持 PiPNN。
+上表的 `pipnn_*` 参数用于调节该构建器，`ef_construction` 不生效；
 现有的 `alpha` 参数控制 PiPNN 的最终 robust pruning。
 
 `build_thread_count` 会并行化向量预处理、分区、候选边生成和最终剪枝。性能测试时应固定
@@ -136,8 +136,13 @@ PiPNN，并通过查询验证每一份生成的图。
 ### 向量存储去重
 
 同时设置 `support_duplicate: true` 和 `deduplicate_storage: true` 后，重复向量会共享
-同一个物理编码槽位，但仍保留各自的标签。该选项目前仅支持使用 `graph_type: "nsw"` 的
-稠密向量 HGraph 索引；`graph_type: "odescent"` 和 `graph_type: "pipnn"` 不支持。
+同一个物理编码槽位，但仍保留各自的标签。该选项支持使用 `graph_type: "nsw"` 或 `"pipnn"` 的
+稠密向量 HGraph 索引；`graph_type: "odescent"` 不支持。
+
+PiPNN 初始 `Build` 按 FP32 原始输入行的字节相等关系分组，每个代表向量只编码一次，
+各 label 的属性和 extra_info 仍独立保存，也可同时启用动态 alpha。
+`duplicate_distance_threshold` 不参与 PiPNN 批量分组；后续增量 `Add` 沿用现有的
+距离阈值或编码判重规则。量化器仍使用原有训练样本进行训练。
 
 启用存储去重后，暂不支持以下操作和配置：
 

src/algorithm/hgraph/hgraph.cpp

@@ -96,9 +96,10 @@ HGraph::HGraph(const HGraphParameterPtr& hgraph_param, const vsag::IndexCommonPa
                                 "HGraph PiPNN only supports dense float32 indexes");
         }
     }
-    if (this->deduplicate_storage_ && this->graph_type_ != GRAPH_TYPE_VALUE_NSW) {
+    if (this->deduplicate_storage_ && this->graph_type_ != GRAPH_TYPE_VALUE_NSW &&
+        this->graph_type_ != GRAPH_TYPE_VALUE_PIPNN) {
         throw VsagException(ErrorType::INVALID_ARGUMENT,
-                            "HGraph deduplicate_storage only supports nsw graph");
+                            "HGraph deduplicate_storage only supports nsw or pipnn graph");
     }
     this->persist_source_id_ = hgraph_param->persist_source_id;
     if (this->using_dedup_storage()) {
@@ -149,7 +150,6 @@ HGraph::HGraph(const HGraphParameterPtr& hgraph_param, const vsag::IndexCommonPa
     mult_ = 1 / log(1.0 * static_cast<double>(this->bottom_graph_->MaximumDegree()));
 
     this->resize_increase_count_bit_ = hgraph_param->resize_increase_count_bit;
-    init_resize_bit_and_reorder();
 
     this->parallel_searcher_ =
         std::make_shared<ParallelSearcher>(common_param, thread_pool_, neighbors_mutex_);
@@ -170,11 +170,17 @@ HGraph::HGraph(const HGraphParameterPtr& hgraph_param, const vsag::IndexCommonPa
             this->high_precise_codes_ = MakeCodeSlotFlattenAdapter(
                 this->high_precise_codes_, this->code_slot_map_, allocator_, &this->total_count_);
         }
+        if (this->raw_vector_ != nullptr and not this->create_new_raw_vector_) {
+            // Shared raw storage must keep the logical-ID adapter too.
+            this->check_and_init_raw_vector(hgraph_param->raw_vector_param, common_param, false);
+        }
         if (this->create_new_raw_vector_ && this->raw_vector_ != nullptr) {
             this->raw_vector_ = MakeCodeSlotFlattenAdapter(
                 this->raw_vector_, this->code_slot_map_, allocator_, &this->total_count_);
         }
     }
+    // Reorder retains its flatten pointer, so initialize it after slot wrapping.
+    init_resize_bit_and_reorder();
     resize(bottom_graph_->max_capacity_);
 }
 

src/algorithm/hgraph/hgraph.h

@@ -469,6 +469,12 @@ class HGraph : public InnerIndexInterface {
     std::vector<int64_t>
     build_by_batch_graph(const DatasetPtr& data, bool use_pipnn);
 
+    /// Encode each PiPNN representative once, then bind all logical IDs to its slot.
+    void
+    materialize_pipnn_storage(const DatasetPtr& data,
+                              const Vector<InnerIdType>& inner_ids,
+                              const Vector<int64_t>& valid_indices);
+
     /// Write codes for inner_id into the persistent flatten storage.
     void
     insert_persistent_codes(const void* data,

src/algorithm/hgraph/hgraph_add_test.cpp

@@ -1272,6 +1272,7 @@ TEST_CASE("HGraph deduplicate_storage batch Add supports internal parallel add",
     constexpr uint64_t build_threads = 4;
     auto common_param = MakeCommonParam(dim, build_threads);
     auto hgraph_json = MakeFp32HGraphJson(true, static_cast<int64_t>(build_threads));
+    hgraph_json["graph_type"].SetString(GENERATE("nsw", "pipnn"));
     auto index = MakeHGraphIndex(hgraph_json, common_param);
 
     std::vector<float> seed_vectors = {
@@ -1419,6 +1420,7 @@ TEST_CASE("HGraph deduplicate_storage concurrent Add keeps visible duplicates se
     constexpr int64_t dim = 2;
     auto common_param = MakeCommonParam(dim);
     auto hgraph_json = MakeFp32HGraphJson();
+    hgraph_json["graph_type"].SetString(GENERATE("nsw", "pipnn"));
     auto index = MakeHGraphIndex(hgraph_json, common_param);
 
     std::vector<float> seed_vectors = {

src/algorithm/hgraph/hgraph_build.cpp

@@ -321,9 +321,9 @@ HGraph::build_by_batch_graph(const DatasetPtr& data, bool use_pipnn) {
         batch_insert_flatten_codes(temporary_sq8_build_data);
         temporary_codes_batched = true;
     }
-    // Every code page of the batch is written at this point (or by the per-row loop below when
-    // rows were skipped), so the batch can be published exactly like a sequence of Add() calls.
-    this->PublishThroughTotalCount();
+    if (not this->using_dedup_storage()) {
+        this->PublishThroughTotalCount();
+    }
     Vector<std::pair<InnerIdType, int64_t>> deferred_code_ids(allocator_);
     Vector<const float*> pipnn_vectors(allocator_);
     if (use_pipnn) {
@@ -341,7 +341,7 @@ HGraph::build_by_batch_graph(const DatasetPtr& data, bool use_pipnn) {
         }
         if (defer_persistent_codes) {
             deferred_code_ids.emplace_back(inner_id, i);
-        } else if (not persistent_codes_batched) {
+        } else if (not persistent_codes_batched and not this->using_dedup_storage()) {
             this->insert_persistent_codes(get_data(data, i), inner_id);
         }
         if (temporary_sq8_build_data != nullptr and not temporary_codes_batched) {
@@ -406,6 +406,9 @@ HGraph::build_by_batch_graph(const DatasetPtr& data, bool use_pipnn) {
                                         this->thread_pool_.get(),
                                         this->build_thread_count_);
         pipnn_builder.Build(bottom_graph_, inner_ids, pipnn_vectors);
+        if (this->using_dedup_storage() and not defer_persistent_codes) {
+            this->materialize_pipnn_storage(data, inner_ids, valid_indices);
+        }
         if (this->support_duplicate_) {
             entry_point_id_ = bottom_graph_->GetGroupId(entry_point_id_);
             for (auto& route_graph_id : route_graph_ids) {
@@ -458,17 +461,92 @@ HGraph::build_by_batch_graph(const DatasetPtr& data, bool use_pipnn) {
         if (this->rabitq_fused_datacell_ == nullptr) {
             this->train_codes_with_dataset(this->sample_train_dataset(data));
         }
-        if (all_rows_are_valid) {
+        if (this->using_dedup_storage()) {
+            this->materialize_pipnn_storage(data, inner_ids, valid_indices);
+        } else if (all_rows_are_valid) {
             batch_insert_persistent_codes();
         } else {
             for (const auto& [inner_id, local_idx] : deferred_code_ids) {
                 this->insert_persistent_codes(get_data(data, local_idx), inner_id);
             }
         }
     }
+    if (this->using_dedup_storage()) {
+        // All logical IDs now have complete codes and a published slot mapping.
+        this->PublishThroughTotalCount();
+    }
     return failed_ids;
 }
 
+void
+HGraph::materialize_pipnn_storage(const DatasetPtr& data,
+                                  const Vector<InnerIdType>& inner_ids,
+                                  const Vector<int64_t>& valid_indices) {
+    struct Representative {
+        InnerIdType inner_id;
+        CodeSlotIdType slot;
+        int64_t row;
+    };
+    Vector<Representative> representatives(allocator_);
+    for (uint64_t i = 0; i < inner_ids.size(); ++i) {
+        const auto id = inner_ids[i];
+        if (bottom_graph_->GetGroupId(id) == id) {
+            representatives.push_back({id, code_slot_map_->AllocateSlot(), valid_indices[i]});
+        }
+    }
+    this->ensure_physical_code_capacity(code_slot_map_->PhysicalCount());
+
+    // Capacity is fixed before any worker writes. Drain every task before publishing mappings
+    // or releasing the representative rows, including when scheduling or encoding throws.
+    const uint64_t workers = std::min<uint64_t>(
+        representatives.size(),
+        thread_pool_ != nullptr ? std::max<uint64_t>(1, build_thread_count_) : 1);
+    std::vector<std::future<void>> futures;
+    futures.reserve(workers);
+    std::exception_ptr failure;
+    try {
+        for (uint64_t worker = 0; worker < workers; ++worker) {
+            const auto begin = representatives.size() * worker / workers;
+            const auto end = representatives.size() * (worker + 1) / workers;
+            auto encode = [this, &representatives, &data, begin, end]() {
+                for (auto i = begin; i < end; ++i) {
+                    const auto& representative = representatives[i];
+                    this->insert_persistent_codes_to_slot(get_data(data, representative.row),
+                                                          representative.slot);
+                }
+            };
+            if (workers > 1) {
+                futures.emplace_back(thread_pool_->GeneralEnqueue(std::move(encode)));
+            } else {
+                encode();
+            }
+        }
+    } catch (...) {
+        failure = std::current_exception();
+    }
+    for (auto& future : futures) {
+        try {
+            future.get();
+        } catch (...) {
+            if (failure == nullptr) {
+                failure = std::current_exception();
+            }
+        }
+    }
+    if (failure != nullptr) {
+        std::rethrow_exception(failure);
+    }
+    for (const auto& representative : representatives) {
+        code_slot_map_->PublishSlot(representative.inner_id, representative.slot);
+    }
+    for (auto id : inner_ids) {
+        const auto group = bottom_graph_->GetGroupId(id);
+        if (group != id) {
+            code_slot_map_->PublishSlot(id, code_slot_map_->Resolve(group));
+        }
+    }
+}
+
 std::vector<int64_t>
 HGraph::Add(const DatasetPtr& data) {
     if (graph_type_ == GRAPH_TYPE_VALUE_PIPNN) {