feat: search_iterator over emb_list (ArrayOfVector) fields

docs/design-docs/design_docs/20260608-emblist_search_iterator.md

@@ -0,0 +1,128 @@
+# MEP: search_iterator() over emb_list (ArrayOfVector), hybrid, and sparse
+
+Tracking discussion: milvus-io/milvus#49906
+
+## Summary (required)
+
+Milvus v2.6 added `emb_list` / `ArrayOfVector` fields with `MAX_SIM` scoring (#42148,
+#43726), enabling ColBERT-style late-interaction retrieval where one row (a document or
+chunk) carries an array of vectors. `search()` over `emb_list` works; `search_iterator()`
+does **not** — nor does it work for hybrid search or stream over sparse vectors. This MEP
+extends Milvus's existing stateless **Iterator v2** path (plus a knowhere change and
+client-side RRF fusion in the SDKs) to close that gap, so deep chunk-level retrieval runs
+in **bounded memory**.
+
+## Motivation (required)
+
+Late-interaction scoring is `MAX_SIM(Q, row) = Σ_i max_{p ∈ row} sim(q_i, p)` — a
+sum-of-max over a *list* of query vectors against a *list* of stored vectors. Three gaps
+compound to block iteration today:
+
+1. **No iterator for `emb_list`.** `MAX_SIM` is a sum-of-max row score, not a
+   single-vector distance, so there is no monotonic radius/`last_bound` cursor of the
+   kind the iterator path filters on for ordinary dense vectors.
+2. **No iterator for hybrid search.** Reciprocal Rank Fusion needs each modality's
+   global rank; a scalar `last_bound` cursor cannot carry it.
+3. **No streaming iterator for sparse.** Today's sparse `AnnIterator` materializes all
+   distances on the first `Next()` (a full index scan), defeating bounded-memory deep
+   retrieval.
+
+Consequence: retrieving beyond the per-search `topK` ceiling forces operators to raise
+`topKLimit` / `maxQueryResultWindow` and run a full deep search at once; peak memory
+then scales with depth × concurrency. A streaming iterator runs deep retrieval in
+bounded memory and removes that ceiling.
+
+## Public Interfaces (optional)
+
+- `search_iterator()` / `searchIterator` accept an `emb_list` query (one query's
+  multiple vectors, passed as an `EmbeddingList`) over an `emb_list` field.
+- A client-side **hybrid** `search_iterator` / `hybridSearchIterator` that fuses
+  dense + emb_list (or sparse) modalities via RRF over the streamed per-modality
+  results. No new server proto/RPC.
+
+## Design Details (required)
+
+The design extends the existing **stateless Iterator v2** rather than building a
+parallel stateful iterator. Each `next()` is an ordinary `Search` RPC carrying a
+`last_bound` cursor; the proxy holds no per-iterator state and query nodes retain
+nothing between batches, so server-side working-set memory is flat in the total result
+depth `L`. (Cost is super-linear in `L` — each batch rebuilds the iterator and
+re-traverses seen results — a known v1 limitation; see *Compatibility* and the v2 note.)
+
+**knowhere**
+- *`AnnIterator` for the `emb_list` HNSW index.* The index is an HNSW graph over the
+  concatenated paragraph vectors of all rows in a segment plus an offset map (which
+  vectors belong to which row). The existing `AnnIterator` returns `m` per-query-vector
+  streaming HNSW sub-iterators; the new emblist iterator is a thin **grouping layer**
+  over them — it advances the sub-iterators, resolves each touched paragraph to its row,
+  computes the **exact** `MAX_SIM` for a row on first sighting (its paragraphs are
+  contiguous and few), and emits row-level `(id, max_sim)` in approximately-descending
+  order under a soft upper bound. It composes HNSW traversal, not reimplements it.
+- *Streaming bounded-WAND iterator for sparse / BM25.* Each `Next(batch)` runs a
+  WAND/MaxScore top-`batch` retrieval bounded above by the previous batch's minimum
+  `(score, id)` cursor (Zipfian BM25 postings keep WAND pruning effective).
+
+**milvus core / segcore** — `CachedSearchIterator` already wraps a knowhere
+`AnnIterator`. Changes: treat `nq` for an emblist query as the number of query
+emb_lists; handle row-level `(id, max_sim)` results; lift the iterator path's
+rejection of `ArrayOfVector`. The brute-force / growing-segment iterator paths (which
+have no emblist `AnnIterator` yet) return a clean, typed `Unsupported` error rather than
+an internal assertion. No new CGO entry points, no iterator registry, no new proto.
+
+**milvus proxy** — the Iterator v2 proxy path is already stateless; it only stops
+rejecting `ArrayOfVector` for the iterator.
+
+**SDKs** — hybrid iteration is **client-side**: the SDK drives two stateless
+single-modality `search_iterator`s and fuses their score-descending streams via RRF
+(NRA threshold algorithm), pinning one MVCC snapshot for the whole hybrid iteration.
+Because the SDK sees both streams it sees each modality's global rank, which a
+server-side scalar cursor cannot. The in-flight (seen-but-not-emitted) map is bounded
+by stream skew.
+
+## Compatibility, Deprecation, and Migration Plan (optional)
+
+Purely additive — no change to existing `search()` / `search_iterator()` over ordinary
+vectors, no proto/RPC changes, no schema migration. Existing `emb_list` `search()` is
+unaffected. **v1 scope** covers the **sealed vector-index path** with
+**restart-on-error** failure handling (a failed batch is an ordinary `Search` failure;
+the caller retries the batch / restarts from the last `last_bound` — there is no
+server-side session to lose). **v2** (deferred): a resumable retained-cursor iterator
+making deep retrieval linear-cost, transparent mid-iteration re-seed, and emblist
+iteration over brute-force / growing-segment paths (blocked on knowhere
+`BruteForce::AnnIterator` gaining emblist support).
+
+## Test Plan (required)
+
+Validated end-to-end on a v2.6.18 cluster against an exact brute-force `MAX_SIM` oracle,
+on real pre-embedded Wikipedia data (1024-d, paragraphs grouped into articles):
+
+- **Set recall** (the sub-iterator-grouping concern from #49906): on 150k articles /
+  ~828k paragraphs, recall@100 = 1.000, recall@1000 = 0.997, ordering Spearman ρ = 1.000
+  (top-100). Held at 10× scale (15k → 150k) — recall@L is bounded by the HNSW index's
+  own ef-limited recall, not the grouping. An **adversarial** set built to trigger the
+  exact failure mode (high-`MAX_SIM` rows whose chunks are buried behind decoys in every
+  sub-iterator): 50/50 surfaced in the exact oracle top-50.
+- **No duplicates / score-exactness / deeper L**: verified by full-iteration comparison
+  to the oracle.
+- **Bounded memory**: iterating 40k hits, client RSS stayed flat (+0.7 MB) and QueryNode
+  RSS stayed bounded (no growth with depth).
+- **Concurrency**: 32 concurrent deep iterators completed, zero OOMKilled, QueryNode
+  memory bounded.
+- **Unit/regression**: knowhere UT suite; milvus segcore C++ + proxy Go tests; pymilvus
+  full suite (zero regressions); milvus-sdk-node spec — all green.
+
+## Rejected Alternatives (optional)
+
+- **A stateful, server-side iterator with a coordinating delegator / session.** Rejected:
+  it would add a per-iterator session (placement, TTL, memory, churn-recovery) and is
+  fragile under node churn. The stateless Iterator-v2 model needs none of that — the
+  cursor lives in the client — and global hybrid rank is recovered by fusing client-side.
+- **Server-side hybrid fusion.** Rejected: a server scalar cursor cannot carry each
+  modality's global rank; client-side RRF over the two streams can.
+
+## References (optional)
+
+- Tracking issue: milvus-io/milvus#49906
+- emb_list / ArrayOfVector + MAX_SIM: #42148, #43726
+- ColBERT / late interaction: Khattab & Zaharia (SIGIR'20); ColBERTv2 (NAACL'22)
+- NRA / threshold algorithm: Fagin, Lotem, Naor (2003)

internal/core/src/query/CachedSearchIterator.cpp

@@ -11,6 +11,7 @@
 
 #include <algorithm>
 
+#include "knowhere/comp/index_param.h"
 #include "query/CachedSearchIterator.h"
 #include "query/SearchBruteForce.h"
 
@@ -27,6 +28,21 @@ CachedSearchIterator::CachedSearchIterator(
                   "Query dataset is nullptr, cannot initialize iterator");
     }
     nq_ = query_ds->GetRows();
+    // For an embedding-list (MAX_SIM) query the dataset is a flat run of
+    // vectors grouped into emb_lists by EMB_LIST_OFFSET; the iterator's unit of
+    // work is the query emb_list, not the individual vector. knowhere's emblist
+    // AnnIterator returns one iterator per query emb_list, so nq is the emb_list
+    // count -- the offset array's index at which it reaches the vector count.
+    const auto* el_offsets =
+        query_ds->Get<const size_t*>(knowhere::meta::EMB_LIST_OFFSET);
+    if (el_offsets != nullptr) {
+        const auto num_vectors = static_cast<size_t>(query_ds->GetRows());
+        size_t num_emb_lists = 0;
+        while (el_offsets[num_emb_lists] < num_vectors) {
+            ++num_emb_lists;
+        }
+        nq_ = num_emb_lists;
+    }
     Init(search_info);
 
     auto search_json = index.PrepareSearchParams(search_info);

internal/core/src/query/CachedSearchIteratorTest.cpp

@@ -10,6 +10,7 @@
 // or implied. See the License for the specific language governing permissions and limitations under the License
 
 #include <gtest/gtest.h>
+#include <limits>
 #include <memory>
 #include <random>
 #include <unordered_set>
@@ -21,6 +22,7 @@
 #include "index/Index.h"
 #include "knowhere/comp/index_param.h"
 #include "query/CachedSearchIterator.h"
+#include "query/SearchBruteForce.h"
 #include "index/VectorIndex.h"
 #include "index/IndexFactory.h"
 #include "knowhere/dataset.h"
@@ -926,3 +928,172 @@ INSTANTIATE_TEST_SUITE_P(
         }
         return constructor_type_str;
     });
+
+// PR 3 — emblist (MAX_SIM) iterator enablement.
+// Before the nq_ fix in the sealed-index constructor, an emblist query (whose
+// dataset is a flat run of vectors with EMB_LIST_OFFSET) made nq_ = the vector
+// count, so CachedSearchIterator::Init threw "Number of queries is greater than
+// 1". nq_ is now the query emb_list count, matching the iterators knowhere's
+// emblist AnnIterator returns.
+TEST(CachedSearchIteratorEmbListTest, EmbListNextBatch) {
+    constexpr int64_t kElDim = 16;
+    constexpr int64_t kNumEmbLists = 200;
+    constexpr int64_t kElBatch = 50;
+    const MetricType metric = knowhere::metric::MAX_SIM_COSINE;
+
+    // variable-length emb_lists (1..5 vectors each) exercise the offset logic
+    std::vector<size_t> offsets{0};
+    for (int64_t i = 0; i < kNumEmbLists; ++i) {
+        offsets.push_back(offsets.back() + static_cast<size_t>(i % 5) + 1);
+    }
+    const int64_t total_vectors = static_cast<int64_t>(offsets.back());
+
+    std::mt19937 rng(42);
+    std::uniform_real_distribution<float> dist(-1.0f, 1.0f);
+    std::vector<float> base(total_vectors * kElDim);
+    for (auto& v : base) {
+        v = dist(rng);
+    }
+
+    auto build_ds = knowhere::GenDataSet(total_vectors, kElDim, base.data());
+    build_ds->Set(knowhere::meta::EMB_LIST_OFFSET,
+                  const_cast<const size_t*>(offsets.data()));
+
+    milvus::index::CreateIndexInfo create_index_info;
+    create_index_info.field_type = DataType::VECTOR_ARRAY;
+    create_index_info.metric_type = metric;
+    create_index_info.index_type = knowhere::IndexEnum::INDEX_HNSW;
+    create_index_info.index_engine_version =
+        knowhere::Version::GetCurrentVersion().VersionNumber();
+    // a VECTOR_ARRAY (emb_list) index dispatches on the inner element type,
+    // which IndexFactory reads from the field schema
+    milvus::storage::FileManagerContext file_manager_context;
+    file_manager_context.fieldDataMeta.field_schema.set_data_type(
+        static_cast<proto::schema::DataType>(DataType::VECTOR_ARRAY));
+    file_manager_context.fieldDataMeta.field_schema.set_element_type(
+        static_cast<proto::schema::DataType>(DataType::VECTOR_FLOAT));
+    auto index = milvus::index::IndexFactory::GetInstance().CreateIndex(
+        create_index_info, file_manager_context);
+    auto build_conf = knowhere::Json{
+        {knowhere::meta::METRIC_TYPE, metric},
+        {knowhere::meta::DIM, std::to_string(kElDim)},
+        {knowhere::indexparam::M, std::to_string(24)},
+        {knowhere::indexparam::EFCONSTRUCTION, std::to_string(360)}};
+    index->BuildWithDataset(build_ds, build_conf);
+    auto* vec_index = dynamic_cast<milvus::index::VectorIndex*>(index.get());
+    ASSERT_NE(vec_index, nullptr);
+
+    // a single query emb_list of 4 vectors
+    std::vector<float> q(4 * kElDim);
+    for (auto& v : q) {
+        v = dist(rng);
+    }
+    auto query_ds = knowhere::GenDataSet(4, kElDim, q.data());
+    std::vector<size_t> q_offsets{0, 4};
+    query_ds->Set(knowhere::meta::EMB_LIST_OFFSET,
+                  const_cast<const size_t*>(q_offsets.data()));
+
+    SearchInfo search_info;
+    search_info.topk_ = kElBatch;
+    search_info.round_decimal_ = -1;
+    search_info.metric_type_ = metric;
+    search_info.search_params_ = {
+        {knowhere::indexparam::EF, std::to_string(128)}};
+    search_info.iterator_v2_info_ =
+        SearchIteratorV2Info{.batch_size = kElBatch};
+
+    auto iterator = std::make_unique<CachedSearchIterator>(
+        *vec_index, query_ds, search_info, nullptr);
+    SearchResult search_result;
+    iterator->NextBatch(search_info, search_result);
+
+    // one query emb_list -> one result row of kElBatch chunk-level results
+    EXPECT_EQ(search_result.total_nq_, 1);
+    EXPECT_EQ(search_result.seg_offsets_.size(), kElBatch);
+    EXPECT_EQ(search_result.distances_.size(), kElBatch);
+
+    // emitted ids are valid chunk (emb_list) ids, deduplicated, descending score
+    std::unordered_set<int64_t> seen;
+    float prev = std::numeric_limits<float>::max();
+    size_t emitted = 0;
+    for (int64_t i = 0; i < kElBatch; ++i) {
+        const auto id = search_result.seg_offsets_[i];
+        if (id == -1) {
+            continue;  // batch padding
+        }
+        EXPECT_GE(id, 0);
+        EXPECT_LT(id, kNumEmbLists);
+        EXPECT_TRUE(seen.insert(id).second) << "duplicate chunk id " << id;
+        EXPECT_LE(search_result.distances_[i], prev);  // MAX_SIM: descending
+        prev = search_result.distances_[i];
+        ++emitted;
+    }
+    EXPECT_GT(emitted, 0u);
+
+    // a second batch, bounded by the first batch's worst score, must not repeat
+    search_info.iterator_v2_info_->last_bound = prev;
+    SearchResult second;
+    iterator->NextBatch(search_info, second);
+    for (int64_t i = 0; i < kElBatch; ++i) {
+        const auto id = second.seg_offsets_[i];
+        if (id == -1) {
+            continue;
+        }
+        EXPECT_TRUE(seen.insert(id).second)
+            << "chunk id " << id << " repeated across batches";
+    }
+}
+
+// R9 / #15: emb_list (VECTOR_ARRAY) search_iterator is supported only on the sealed
+// vector-index path (knowhere's emblist AnnIterator, exercised by EmbListNextBatch
+// above). The brute-force / growing-segment iterator path has no emblist support, so
+// it must fail with a clean, typed Unsupported error -- not a bare assertion failure
+// deep in segcore. This guards the graceful behaviour lifted into the proxy by PR 4.
+TEST(CachedSearchIteratorEmbListTest, BruteForceIteratorRejectsEmbListGracefully) {
+    constexpr int64_t kDim = 4;
+    // one emb_list of two vectors; raw/query data is a flat run of vectors keyed by
+    // EMB_LIST_OFFSET (offsets length = num emb_lists + 1).
+    std::vector<float> base(2 * kDim, 0.1f);
+    std::vector<size_t> base_offsets{0, 2};
+    std::vector<float> query(2 * kDim, 0.2f);
+    std::vector<size_t> query_offsets{0, 2};
+
+    dataset::RawDataset raw_ds{
+        .dim = kDim,
+        .num_raw_data = 1,
+        .raw_data = base.data(),
+        .raw_data_offsets = base_offsets.data(),
+    };
+    dataset::SearchDataset query_ds{
+        .metric_type = knowhere::metric::MAX_SIM_COSINE,
+        .num_queries = 1,
+        .topk = 10,
+        .round_decimal = -1,
+        .dim = kDim,
+        .query_data = query.data(),
+        .query_offsets = query_offsets.data(),
+    };
+    SearchInfo search_info{
+        .topk_ = 10,
+        .round_decimal_ = -1,
+        .metric_type_ = knowhere::metric::MAX_SIM_COSINE,
+        .iterator_v2_info_ = SearchIteratorV2Info{.batch_size = 10},
+    };
+    std::map<std::string, std::string> index_info;
+    BitsetView bitset;
+
+    try {
+        GetBruteForceSearchIterators(query_ds,
+                                     raw_ds,
+                                     search_info,
+                                     index_info,
+                                     bitset,
+                                     DataType::VECTOR_ARRAY);
+        FAIL() << "expected emb_list brute-force search_iterator to be rejected";
+    } catch (const SegcoreError& e) {
+        EXPECT_EQ(e.get_error_code(), ErrorCode::Unsupported);
+        EXPECT_NE(std::string(e.what()).find("brute-force / growing"),
+                  std::string::npos)
+            << "unexpected error message: " << e.what();
+    }
+}

internal/core/src/query/SearchBruteForce.cpp

@@ -278,8 +278,18 @@ DispatchBruteForceIteratorByDataType(const knowhere::DataSetPtr& base_dataset,
                                      const knowhere::Json& config,
                                      const BitsetView& bitset,
                                      milvus::DataType data_type) {
-    AssertInfo(data_type != DataType::VECTOR_ARRAY,
-               "VECTOR_ARRAY is not supported for brute force iterator");
+    // emb_list (VECTOR_ARRAY) search_iterator is supported only on the sealed
+    // vector-index path (knowhere's emblist AnnIterator). The brute-force /
+    // growing-segment iterator path does not support it yet (R9, deferred to v2):
+    // knowhere's BruteForce::AnnIterator has no emb_list overload. Surface a clean,
+    // typed Unsupported error here rather than tripping the bare assert below, so a
+    // search_iterator over a collection with growing / un-indexed segments returns a
+    // graceful "not supported" instead of an opaque internal assertion failure.
+    if (data_type == DataType::VECTOR_ARRAY) {
+        ThrowInfo(ErrorCode::Unsupported,
+                  "search_iterator over emb_list (vector array) fields is not "
+                  "supported on brute-force / growing segments");
+    }
 
     switch (data_type) {
         case DataType::VECTOR_FLOAT: