fix: ensure deleted chunks are not returned in retrieval (#12520)

api/apps/chunk_app.py

@@ -223,7 +223,9 @@ def _rm_sync():
             e, doc = DocumentService.get_by_id(req["doc_id"])
             if not e:
                 return get_data_error_result(message="Document not found!")
-            if not settings.docStoreConn.delete({"id": req["chunk_ids"]},
+            # Include doc_id in condition to properly scope the delete
+            condition = {"id": req["chunk_ids"], "doc_id": req["doc_id"]}
+            if not settings.docStoreConn.delete(condition,
                                                 search.index_name(DocumentService.get_tenant_id(req["doc_id"])),
                                                 doc.kb_id):
                 return get_data_error_result(message="Chunk deleting failure")

api/db/services/document_service.py

@@ -340,14 +340,35 @@ def insert(cls, doc):
     def remove_document(cls, doc, tenant_id):
         from api.db.services.task_service import TaskService
         cls.clear_chunk_num(doc.id)
+
+        # Delete tasks first
         try:
             TaskService.filter_delete([Task.doc_id == doc.id])
+        except Exception as e:
+            logging.warning(f"Failed to delete tasks for document {doc.id}: {e}")
+
+        # Delete chunk images (non-critical, log and continue)
+        try:
             cls.delete_chunk_images(doc, tenant_id)
+        except Exception as e:
+            logging.warning(f"Failed to delete chunk images for document {doc.id}: {e}")
+
+        # Delete thumbnail (non-critical, log and continue)
+        try:
             if doc.thumbnail and not doc.thumbnail.startswith(IMG_BASE64_PREFIX):
                 if settings.STORAGE_IMPL.obj_exist(doc.kb_id, doc.thumbnail):
                     settings.STORAGE_IMPL.rm(doc.kb_id, doc.thumbnail)
+        except Exception as e:
+            logging.warning(f"Failed to delete thumbnail for document {doc.id}: {e}")
+
+        # Delete chunks from doc store - this is critical, log errors
+        try:
             settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
+        except Exception as e:
+            logging.error(f"Failed to delete chunks from doc store for document {doc.id}: {e}")
 
+        # Cleanup knowledge graph references (non-critical, log and continue)
+        try:
             graph_source = settings.docStoreConn.get_fields(
                 settings.docStoreConn.search(["source_id"], [], {"kb_id": doc.kb_id, "knowledge_graph_kwd": ["graph"]}, [], OrderByExpr(), 0, 1, search.index_name(tenant_id), [doc.kb_id]), ["source_id"]
             )
@@ -360,8 +381,9 @@ def remove_document(cls, doc, tenant_id):
                                              search.index_name(tenant_id), doc.kb_id)
                 settings.docStoreConn.delete({"kb_id": doc.kb_id, "knowledge_graph_kwd": ["entity", "relation", "graph", "subgraph", "community_report"], "must_not": {"exists": "source_id"}},
                                              search.index_name(tenant_id), doc.kb_id)
-        except Exception:
-            pass
+        except Exception as e:
+            logging.warning(f"Failed to cleanup knowledge graph for document {doc.id}: {e}")
+
         return cls.delete_by_id(doc.id)
 
     @classmethod

rag/utils/es_conn.py

@@ -303,32 +303,43 @@ def update(self, condition: dict, new_value: dict, index_name: str, knowledgebas
     def delete(self, condition: dict, index_name: str, knowledgebase_id: str) -> int:
         assert "_id" not in condition
         condition["kb_id"] = knowledgebase_id
+
+        # Build a bool query that combines id filter with other conditions
+        bool_query = Q("bool")
+
+        # Handle chunk IDs if present
         if "id" in condition:
             chunk_ids = condition["id"]
             if not isinstance(chunk_ids, list):
                 chunk_ids = [chunk_ids]
-            if not chunk_ids:  # when chunk_ids is empty, delete all
-                qry = Q("match_all")
-            else:
-                qry = Q("ids", values=chunk_ids)
-        else:
-            qry = Q("bool")
-            for k, v in condition.items():
-                if k == "exists":
-                    qry.filter.append(Q("exists", field=v))
+            if chunk_ids:
+                # Filter by specific chunk IDs
+                bool_query.filter.append(Q("ids", values=chunk_ids))
+            # If chunk_ids is empty, we don't add an ids filter - rely on other conditions
 
-                elif k == "must_not":
-                    if isinstance(v, dict):
-                        for kk, vv in v.items():
-                            if kk == "exists":
-                                qry.must_not.append(Q("exists", field=vv))
+        # Add all other conditions as filters
+        for k, v in condition.items():
+            if k == "id":
+                continue  # Already handled above
+            if k == "exists":
+                bool_query.filter.append(Q("exists", field=v))
+            elif k == "must_not":
+                if isinstance(v, dict):
+                    for kk, vv in v.items():
+                        if kk == "exists":
+                            bool_query.must_not.append(Q("exists", field=vv))
+            elif isinstance(v, list):
+                bool_query.must.append(Q("terms", **{k: v}))
+            elif isinstance(v, str) or isinstance(v, int):
+                bool_query.must.append(Q("term", **{k: v}))
+            elif v is not None:
+                raise Exception("Condition value must be int, str or list.")
 
-                elif isinstance(v, list):
-                    qry.must.append(Q("terms", **{k: v}))
-                elif isinstance(v, str) or isinstance(v, int):
-                    qry.must.append(Q("term", **{k: v}))
-                else:
-                    raise Exception("Condition value must be int, str or list.")
+        # If no filters were added, use match_all (for tenant-wide operations)
+        if not bool_query.filter and not bool_query.must and not bool_query.must_not:
+            qry = Q("match_all")
+        else:
+            qry = bool_query
         self.logger.debug("ESConnection.delete query: " + json.dumps(qry.to_dict()))
         for _ in range(ATTEMPT_TIME):
             try:

rag/utils/opensearch_conn.py

@@ -405,34 +405,45 @@ def update(self, condition: dict, newValue: dict, indexName: str, knowledgebaseI
         return False
 
     def delete(self, condition: dict, indexName: str, knowledgebaseId: str) -> int:
-        qry = None
         assert "_id" not in condition
+        condition["kb_id"] = knowledgebaseId
+
+        # Build a bool query that combines id filter with other conditions
+        bool_query = Q("bool")
+
+        # Handle chunk IDs if present
         if "id" in condition:
             chunk_ids = condition["id"]
             if not isinstance(chunk_ids, list):
                 chunk_ids = [chunk_ids]
-            if not chunk_ids:  # when chunk_ids is empty, delete all
-                qry = Q("match_all")
-            else:
-                qry = Q("ids", values=chunk_ids)
+            if chunk_ids:
+                # Filter by specific chunk IDs
+                bool_query.filter.append(Q("ids", values=chunk_ids))
+            # If chunk_ids is empty, we don't add an ids filter - rely on other conditions
+
+        # Add all other conditions as filters
+        for k, v in condition.items():
+            if k == "id":
+                continue  # Already handled above
+            if k == "exists":
+                bool_query.filter.append(Q("exists", field=v))
+            elif k == "must_not":
+                if isinstance(v, dict):
+                    for kk, vv in v.items():
+                        if kk == "exists":
+                            bool_query.must_not.append(Q("exists", field=vv))
+            elif isinstance(v, list):
+                bool_query.must.append(Q("terms", **{k: v}))
+            elif isinstance(v, str) or isinstance(v, int):
+                bool_query.must.append(Q("term", **{k: v}))
+            elif v is not None:
+                raise Exception("Condition value must be int, str or list.")
+
+        # If no filters were added, use match_all (for tenant-wide operations)
+        if not bool_query.filter and not bool_query.must and not bool_query.must_not:
+            qry = Q("match_all")
         else:
-            qry = Q("bool")
-            for k, v in condition.items():
-                if k == "exists":
-                    qry.filter.append(Q("exists", field=v))
-
-                elif k == "must_not":
-                    if isinstance(v, dict):
-                        for kk, vv in v.items():
-                            if kk == "exists":
-                                qry.must_not.append(Q("exists", field=vv))
-
-                elif isinstance(v, list):
-                    qry.must.append(Q("terms", **{k: v}))
-                elif isinstance(v, str) or isinstance(v, int):
-                    qry.must.append(Q("term", **{k: v}))
-                else:
-                    raise Exception("Condition value must be int, str or list.")
+            qry = bool_query
         logger.debug("OSConnection.delete query: " + json.dumps(qry.to_dict()))
         for _ in range(ATTEMPT_TIME):
             try:

test/testcases/test_http_api/test_chunk_management_within_dataset/test_retrieval_chunks.py

@@ -15,9 +15,10 @@
 #
 import os
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from time import sleep
 
 import pytest
-from common import retrieval_chunks
+from common import add_chunk, delete_chunks, retrieval_chunks
 from configs import INVALID_API_TOKEN
 from libs.auth import RAGFlowHttpApiAuth
 
@@ -310,3 +311,89 @@ def test_concurrent_retrieval(self, HttpApiAuth, add_chunks):
         responses = list(as_completed(futures))
         assert len(responses) == count, responses
         assert all(future.result()["code"] == 0 for future in futures)
+
+
+class TestDeletedChunksNotRetrievable:
+    """Regression tests for issue #12520: deleted slices should not appear in retrieval/reference."""
+
+    @pytest.mark.p1
+    def test_deleted_chunk_not_in_retrieval(self, HttpApiAuth, add_document):
+        """
+        Test that a deleted chunk is not returned by the retrieval API.
+
+        Steps:
+        1. Add a chunk with unique content
+        2. Verify the chunk is retrievable
+        3. Delete the chunk
+        4. Verify the chunk is no longer retrievable
+        """
+        dataset_id, document_id = add_document
+
+        # Add a chunk with unique content that we can search for
+        unique_content = "UNIQUE_TEST_CONTENT_12520_REGRESSION"
+        res = add_chunk(HttpApiAuth, dataset_id, document_id, {"content": unique_content})
+        assert res["code"] == 0, f"Failed to add chunk: {res}"
+        chunk_id = res["data"]["chunk"]["id"]
+
+        # Wait for indexing to complete
+        sleep(2)
+
+        # Verify the chunk is retrievable
+        payload = {"question": unique_content, "dataset_ids": [dataset_id]}
+        res = retrieval_chunks(HttpApiAuth, payload)
+        assert res["code"] == 0, f"Retrieval failed: {res}"
+        chunk_ids_before = [c["id"] for c in res["data"]["chunks"]]
+        assert chunk_id in chunk_ids_before, f"Chunk {chunk_id} should be retrievable before deletion"
+
+        # Delete the chunk
+        res = delete_chunks(HttpApiAuth, dataset_id, document_id, {"chunk_ids": [chunk_id]})
+        assert res["code"] == 0, f"Failed to delete chunk: {res}"
+
+        # Wait for deletion to propagate
+        sleep(1)
+
+        # Verify the chunk is no longer retrievable
+        res = retrieval_chunks(HttpApiAuth, payload)
+        assert res["code"] == 0, f"Retrieval failed after deletion: {res}"
+        chunk_ids_after = [c["id"] for c in res["data"]["chunks"]]
+        assert chunk_id not in chunk_ids_after, f"Chunk {chunk_id} should NOT be retrievable after deletion"
+
+    @pytest.mark.p2
+    def test_deleted_chunks_batch_not_in_retrieval(self, HttpApiAuth, add_document):
+        """
+        Test that multiple deleted chunks are not returned by retrieval.
+        """
+        dataset_id, document_id = add_document
+
+        # Add multiple chunks with unique content
+        chunk_ids = []
+        for i in range(3):
+            unique_content = f"BATCH_DELETE_TEST_CHUNK_{i}_12520"
+            res = add_chunk(HttpApiAuth, dataset_id, document_id, {"content": unique_content})
+            assert res["code"] == 0, f"Failed to add chunk {i}: {res}"
+            chunk_ids.append(res["data"]["chunk"]["id"])
+
+        # Wait for indexing
+        sleep(2)
+
+        # Verify chunks are retrievable
+        payload = {"question": "BATCH_DELETE_TEST_CHUNK", "dataset_ids": [dataset_id]}
+        res = retrieval_chunks(HttpApiAuth, payload)
+        assert res["code"] == 0
+        retrieved_ids_before = [c["id"] for c in res["data"]["chunks"]]
+        for cid in chunk_ids:
+            assert cid in retrieved_ids_before, f"Chunk {cid} should be retrievable before deletion"
+
+        # Delete all chunks
+        res = delete_chunks(HttpApiAuth, dataset_id, document_id, {"chunk_ids": chunk_ids})
+        assert res["code"] == 0, f"Failed to delete chunks: {res}"
+
+        # Wait for deletion to propagate
+        sleep(1)
+
+        # Verify none of the chunks are retrievable
+        res = retrieval_chunks(HttpApiAuth, payload)
+        assert res["code"] == 0
+        retrieved_ids_after = [c["id"] for c in res["data"]["chunks"]]
+        for cid in chunk_ids:
+            assert cid not in retrieved_ids_after, f"Chunk {cid} should NOT be retrievable after deletion"