fix: remove pagerank_fea on ES/OS when chunk feedback hits zero

rank_feature fields must not be set to 0; align chunk feedback with kb_app
by sending remove for elasticsearch/opensearch. Extend single-doc update in
es_conn and opensearch_conn to honor remove alongside doc payloads.

Tests: stub DOC_ENGINE=infinity by default; add ES/OS remove assertions.

Author: MkDev11

api/db/services/chunk_feedback_service.py

@@ -198,9 +198,14 @@ def update_chunk_weight(
             # Clamp to valid range (integer; matches doc-store column types)
             new_weight = max(MIN_PAGERANK_WEIGHT, min(MAX_PAGERANK_WEIGHT, new_weight))
 
-            # Update the chunk
+            # Elasticsearch/OpenSearch map pagerank_fea as rank_feature; zero must not be
+            # indexed — remove the field (same as kb_app / dataset_api_service).
             condition = {"id": chunk_id}
-            new_value = {PAGERANK_FLD: int(new_weight)}
+            engine = settings.DOC_ENGINE.lower()
+            if new_weight == 0 and engine in ("elasticsearch", "opensearch"):
+                new_value = {"remove": PAGERANK_FLD}
+            else:
+                new_value = {PAGERANK_FLD: int(new_weight)}
 
             success = settings.docStoreConn.update(
                 condition, new_value, idx_name, kb_id

rag/utils/es_conn.py

@@ -303,7 +303,9 @@ def update(self, condition: dict, new_value: dict, index_name: str, knowledgebas
             # update specific single document
             chunk_id = condition["id"]
             for i in range(ATTEMPT_TIME):
-                for k in doc.keys():
+                doc_part = copy.deepcopy(doc)
+                remove_field = doc_part.pop("remove", None) if isinstance(doc_part.get("remove"), str) else None
+                for k in doc_part.keys():
                     if "feas" != k.split("_")[-1]:
                         continue
                     try:
@@ -312,8 +314,16 @@ def update(self, condition: dict, new_value: dict, index_name: str, knowledgebas
                         self.logger.exception(
                             f"ESConnection.update(index={index_name}, id={chunk_id}, doc={json.dumps(condition, ensure_ascii=False)}) got exception")
                 try:
-                    self.es.update(index=index_name, id=chunk_id, doc=doc)
-                    return True
+                    if remove_field is not None:
+                        self.es.update(
+                            index=index_name,
+                            id=chunk_id,
+                            script=f"ctx._source.remove('{remove_field}');",
+                        )
+                    if doc_part:
+                        self.es.update(index=index_name, id=chunk_id, doc=doc_part)
+                    if remove_field is not None or doc_part:
+                        return True
                 except Exception as e:
                     self.logger.exception(
                         f"ESConnection.update(index={index_name}, id={chunk_id}, doc={json.dumps(condition, ensure_ascii=False)}) got exception: " + str(

rag/utils/opensearch_conn.py

@@ -329,9 +329,19 @@ def update(self, condition: dict, newValue: dict, indexName: str, knowledgebaseI
             # update specific single document
             chunkId = condition["id"]
             for i in range(ATTEMPT_TIME):
+                doc_part = copy.deepcopy(doc)
+                remove_field = doc_part.pop("remove", None) if isinstance(doc_part.get("remove"), str) else None
                 try:
-                    self.os.update(index=indexName, id=chunkId, body={"doc": doc})
-                    return True
+                    if remove_field is not None:
+                        self.os.update(
+                            index=indexName,
+                            id=chunkId,
+                            body={"script": {"source": f"ctx._source.remove('{remove_field}');"}},
+                        )
+                    if doc_part:
+                        self.os.update(index=indexName, id=chunkId, body={"doc": doc_part})
+                    if remove_field is not None or doc_part:
+                        return True
                 except Exception as e:
                     logger.exception(
                         f"OSConnection.update(index={indexName}, id={id}, doc={json.dumps(condition, ensure_ascii=False)}) got exception")

test/testcases/test_web_api/test_chunk_feedback/test_chunk_feedback_service.py

@@ -43,6 +43,8 @@ def _load_feedback_module(monkeypatch):
 
     settings_mod = ModuleType("common.settings")
     settings_mod.docStoreConn = MagicMock()
+    # Non-ES engines accept pagerank_fea=0; tests below override for elasticsearch/opensearch.
+    settings_mod.DOC_ENGINE = "infinity"
     monkeypatch.setitem(sys.modules, "common.settings", settings_mod)
     common_pkg.settings = settings_mod
 
@@ -251,6 +253,42 @@ def test_update_weight_clamp_min(self, feedback_env):
         new_value = call_args[0][1]
         assert new_value["pagerank_fea"] == mod.MIN_PAGERANK_WEIGHT
 
+    def test_update_weight_zero_elasticsearch_removes_field(self, feedback_env):
+        """rank_feature cannot store 0 on ES — use remove payload (see kb_app)."""
+        mod, settings_mod = feedback_env
+        settings_mod.DOC_ENGINE = "elasticsearch"
+        mock_doc_store = MagicMock()
+        mock_doc_store.get.return_value = {"pagerank_fea": 1}
+        mock_doc_store.update.return_value = True
+        settings_mod.docStoreConn = mock_doc_store
+
+        mod.ChunkFeedbackService.update_chunk_weight(
+            tenant_id="tenant1",
+            chunk_id="chunk1",
+            kb_id="kb1",
+            delta=-1,
+        )
+
+        new_value = mock_doc_store.update.call_args[0][1]
+        assert new_value == {"remove": "pagerank_fea"}
+
+    def test_update_weight_zero_opensearch_removes_field(self, feedback_env):
+        mod, settings_mod = feedback_env
+        settings_mod.DOC_ENGINE = "opensearch"
+        mock_doc_store = MagicMock()
+        mock_doc_store.get.return_value = {"pagerank_fea": 2}
+        mock_doc_store.update.return_value = True
+        settings_mod.docStoreConn = mock_doc_store
+
+        mod.ChunkFeedbackService.update_chunk_weight(
+            tenant_id="tenant1",
+            chunk_id="chunk1",
+            kb_id="kb1",
+            delta=-2,
+        )
+
+        assert mock_doc_store.update.call_args[0][1] == {"remove": "pagerank_fea"}
+
 
 class TestApplyFeedback:
     """Tests for apply_feedback method."""