fix: support Infinity knowledge compilation (#17288)

### What problem does this PR solve?

Fix Infinity compatibility issues in knowledge compilation.

This change:

- Stores compilation source ID lists as JSON arrays in Infinity.
- Parses JSON array fields when reading compiled documents.
- Uses `json_contains` for filtering JSON array fields.
- Adds the missing `name` column to the Infinity mapping.
- Updates dataset navigation KNN search to use the unified
`MatchDenseExpr` interface.
- Handles unavailable embeddings without querying an invalid `q_0_vec`
field.

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)

common/doc_store/infinity_conn_base.py

@@ -304,7 +304,24 @@ class InfinityConnectionBase(DocStoreConnection):
                 continue
             if not v:
                 continue
-            if self.field_keyword(k):
+            if k in {
+                "source_chunk_ids",
+                "source_doc_ids",
+                "compilation_template_ids",
+                "doc_ids_kwd",
+                "entity_names_kwd",
+                "outlinks_kwd",
+                "related_kb_pages_kwd",
+                "rechunked_from_chunk_ids",
+            }:
+                values = v if isinstance(v, list) else [v]
+                json_conditions = []
+                for item in values:
+                    literal = json.dumps(item, ensure_ascii=False).replace("'", "''")
+                    json_conditions.append(f"json_contains({k}, '{literal}')")
+                if json_conditions:
+                    cond.append("(" + " or ".join(json_conditions) + ")")
+            elif self.field_keyword(k):
                 if isinstance(v, list):
                     inCond = list()
                     for item in v:

conf/infinity_mapping.json

@@ -8,6 +8,7 @@
 	"create_timestamp_flt": {"type": "float", "default": 0.0},
 	"img_id": {"type": "varchar", "default": ""},
 	"docnm": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "docnm_kwd, title_tks, title_sm_tks"},
+	"name": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"name_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"tag_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"important_kwd_empty_count": {"type": "integer", "default": 0},
@@ -43,9 +44,9 @@
 	"extra": {"type": "varchar", "default": ""},
 
 	"compile_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"source_chunk_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"source_doc_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"compilation_template_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
+	"source_chunk_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
+	"source_doc_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
+	"compilation_template_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
 	"compilation_template_kind_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"chunk_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"input_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
@@ -55,20 +56,20 @@
 	"skill_with_weight": {"type": "varchar", "default": ""},
 	"skill_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"children_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"doc_ids_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
+	"doc_ids_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
 	"slug_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"title_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"topic_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"page_type_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"entity_names_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"outlinks_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
-	"related_kb_pages_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
+	"entity_names_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
+	"outlinks_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
+	"related_kb_pages_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
 	"type_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"from_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"to_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"rechunk_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
 	"rechunked_from_template_id": {"type": "varchar", "default": ""},
-	"rechunked_from_chunk_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"},
+	"rechunked_from_chunk_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}},
 	"superseded_by_chunk_id": {"type": "varchar", "default": ""},
 	"doc_count_int": {"type": "integer", "default": 0},
 	"depth_int": {"type": "integer", "default": 0},

rag/advanced_rag/knowlege_compile/dataset_nav.py

@@ -177,6 +177,7 @@ async def _store_knn(
 ) -> list[dict]:
     """KNN search with dense vector and filter, returning top_k hits."""
     from common import settings
+    from common.doc_store.doc_store_base import MatchDenseExpr, OrderByExpr
 
     index = _index_name(tenant_id)
     vf = _vec_field(vec_dim)
@@ -191,39 +192,26 @@ async def _store_knn(
         "doc_ids_kwd",
         vf,
     ]
-    try:
-        res = await thread_pool_exec(
-            settings.docStoreConn.search,
-            fields,
-            [],
-            filter_condition,
-            [],
-            None,
-            0,
-            top_k,
-            index,
-            [kb_id],
-            knn_vector=vec,
-            knn_vector_field=vf,
-        )
-    except TypeError:
-        # Fallback: some doc store connectors don't accept knn_* kwargs.
-        # Perform a plain search and lambda-rank in Python (slow-path).
-        rows = await _store_search(
-            tenant_id,
-            kb_id,
-            filter_condition,
-            fields,
-            limit=top_k * 10,
-        )
-        scoring = []
-        for r in rows:
-            stored = r.get(vf)
-            if _vector_len(stored) > 0 and _vector_len(stored) == _vector_len(vec):
-                sim = sum(a * b for a, b in zip(stored, vec))
-                scoring.append((sim, r))
-        scoring.sort(key=lambda x: -x[0])
-        return [r for _, r in scoring[:top_k]]
+    match_expr = MatchDenseExpr(
+        vector_column_name=vf,
+        embedding_data=list(vec),
+        embedding_data_type="float",
+        distance_type="cosine",
+        topn=top_k,
+        extra_options={},
+    )
+    res = await thread_pool_exec(
+        settings.docStoreConn.search,
+        fields,
+        [],
+        filter_condition,
+        [match_expr],
+        OrderByExpr(),
+        0,
+        top_k,
+        index,
+        [kb_id],
+    )
     results = settings.docStoreConn.get_fields(res, fields) if res else {}
     return list(results.values())
 
@@ -590,7 +578,7 @@ async def upsert_dataset_nav_doc(
 
     # 3. Embed doc summary
     doc_embedding = await _embed(embd_mdl, summary) if embd_mdl else []
-    vec_dim = _EMBED_DIM or 0
+    vec_dim = len(doc_embedding)
 
     lock = RedisDistributedLock(
         _nav_lock_key(kb_id),
@@ -605,12 +593,16 @@ async def upsert_dataset_nav_doc(
 
     try:
         # 4. Layered KNN search for nearest cluster
-        best_name, best_parent, sim = await _find_best_cluster(
-            tenant_id,
-            kb_id,
-            doc_embedding,
-            vec_dim,
-        )
+        if doc_embedding:
+            best_name, best_parent, sim = await _find_best_cluster(
+                tenant_id,
+                kb_id,
+                doc_embedding,
+                vec_dim,
+            )
+        else:
+            logging.warning("dataset_nav: embedding unavailable for doc=%s, skipping KNN placement", doc_id)
+            best_name, best_parent, sim = None, None, 0.0
 
         if best_name and sim >= _MERGE_THRESHOLD:
             # ── Merge into best cluster ──

rag/utils/infinity_conn.py

@@ -26,6 +26,20 @@ from common.doc_store.doc_store_base import MatchExpr, MatchTextExpr, MatchDense
 from common.doc_store.infinity_conn_base import InfinityConnectionBase
 
 
+_JSON_LIST_FIELDS = frozenset(
+    (
+        "source_chunk_ids",
+        "source_doc_ids",
+        "compilation_template_ids",
+        "doc_ids_kwd",
+        "entity_names_kwd",
+        "outlinks_kwd",
+        "related_kb_pages_kwd",
+        "rechunked_from_chunk_ids",
+    )
+)
+
+
 @singleton
 class InfinityConnection(InfinityConnectionBase):
     """
@@ -440,6 +454,8 @@ class InfinityConnection(InfinityConnectionBase):
                     elif k == "question_tks":
                         if not d.get("question_kwd"):
                             d["questions"] = self.list2str(v)
+                    elif k in _JSON_LIST_FIELDS:
+                        d[k] = json.dumps(list(v) if isinstance(v, (list, tuple, set)) else [], ensure_ascii=False)
                     elif self.field_keyword(k):
                         if isinstance(v, list):
                             d[k] = "###".join(v)
@@ -579,6 +595,8 @@ class InfinityConnection(InfinityConnectionBase):
                 elif k == "question_tks":
                     if not new_value.get("question_kwd"):
                         new_value["questions"] = self.list2str(v)
+                elif k in _JSON_LIST_FIELDS:
+                    new_value[k] = json.dumps(list(v) if isinstance(v, (list, tuple, set)) else [], ensure_ascii=False)
                 elif self.field_keyword(k):
                     if isinstance(v, list):
                         new_value[k] = "###".join(v)
@@ -794,7 +812,22 @@ class InfinityConnection(InfinityConnectionBase):
 
         for column in list(res2.columns):
             k = column.lower()
-            if self.field_keyword(k):
+            if k in _JSON_LIST_FIELDS:
+
+                def parse_json_list(value):
+                    if isinstance(value, list):
+                        return value
+                    if not value:
+                        return []
+                    try:
+                        parsed = json.loads(value)
+                        return parsed if isinstance(parsed, list) else [parsed]
+                    except (TypeError, json.JSONDecodeError):
+                        # Read rows written by the previous varchar encoding.
+                        return [item for item in str(value).split("###") if item]
+
+                res2[column] = res2[column].apply(parse_json_list)
+            elif self.field_keyword(k):
                 res2[column] = res2[column].apply(lambda v: [kwd for kwd in v.split("###") if kwd])
             elif re.search(r"_feas$", k):
                 res2[column] = res2[column].apply(lambda v: json.loads(v) if v else {})

web/src/hooks/use-document-request.ts

@@ -74,6 +74,8 @@ const DocumentKeys = {
     [DocumentApiAction.FetchDocumentList, 'byIds', ids] as const,
 };
 
+const documentIngestInFlight = new Map<string, Promise<unknown>>();
+
 export const DocumentStructureKeys = {
   graph: (datasetId: string, documentId: string) =>
     [
@@ -389,7 +391,36 @@ export const useRunDocument = () => {
     },
   });
 
-  return { runDocumentByIds: mutateAsync, loading, data };
+  const runDocumentByIds = useCallback(
+    (params: {
+      documentIds: string[];
+      run: number;
+      option?: { delete: boolean; apply_kb: boolean };
+    }) => {
+      const key = JSON.stringify({
+        documentIds: [...params.documentIds].sort(),
+        run: params.run,
+        option: params.option || null,
+      });
+      const existingRequest = documentIngestInFlight.get(key);
+      if (existingRequest) {
+        return existingRequest;
+      }
+
+      const request = mutateAsync(params);
+      documentIngestInFlight.set(key, request);
+      const clearRequest = () => {
+        if (documentIngestInFlight.get(key) === request) {
+          documentIngestInFlight.delete(key);
+        }
+      };
+      void request.then(clearRequest, clearRequest);
+      return request;
+    },
+    [mutateAsync],
+  );
+
+  return { runDocumentByIds, loading, data };
 };
 
 export const useRemoveDocument = () => {