fix: exclude disabled documents from dataset structures (#18041)

api/apps/services/dataset_api_service.py

@@ -607,6 +607,10 @@ def run_index(dataset_id: str, tenant_id: str, index_type: str):
         types=[],
         suffix=[],
     )
+    # Disabled documents must not participate in a dataset-level structure
+    # rebuild. Keep them out of the fan-out task itself; the merger also
+    # applies a database-backed filter as a defense in depth.
+    documents = [document for document in documents if str(document.get("status", "1")) != "0"]
     if not documents:
         return False, f"No documents in Dataset {dataset_id}"
 
@@ -2053,17 +2057,6 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
         except Exception:
             logging.exception("get_dataset_structure: bucket build failed for kb=%s template=%s", dataset_id, tid)
             continue
-        if scope_kwd == "dataset" and not entities and not relations:
-            try:
-                entities, relations = await sgc.build_bucket(
-                    index_nm,
-                    dataset_id,
-                    {"compilation_template_ids": [tid], "scope_kwd": ["doc"], "doc_id": sorted(active_doc_ids)},
-                    excluded_doc_ids=disabled_doc_ids,
-                )
-            except Exception:
-                logging.exception("get_dataset_structure: doc fallback bucket build failed for kb=%s template=%s", dataset_id, tid)
-                continue
         if resolved_kind in {"knowledge_graph", "mind_map", "timeline"}:
             entities = sgc.filter_entities_with_relations(entities, relations)
             if not entities:

api/db/services/document_service.py

@@ -45,7 +45,7 @@ class DocumentService(CommonService):
     @classmethod
     @DB.connection_context()
     def get_disabled_doc_ids_by_kb_id(cls, kb_id) -> set[str]:
-        return {str(doc_id) for doc_id in cls.model.select(cls.model.id).where((cls.model.kb_id == kb_id) & (cls.model.status == "0")).tuples()}
+        return {str(doc_id) for (doc_id,) in cls.model.select(cls.model.id).where((cls.model.kb_id == kb_id) & (cls.model.status == "0")).tuples()}
 
     @classmethod
     def get_cls_model_fields(cls):

rag/svr/task_executor_refactor/dataset_structure_merger.py

@@ -699,24 +699,10 @@ async def _do_build(
         }
         if template_id:
             del_cond["compilation_template_ids"] = [template_id]
-        offset = 0
-        while True:
-            existing = await _index_search(
-                tenant_id,
-                kb_id,
-                del_cond,
-                ["id"],
-                limit=1000,
-                offset=offset,
-            )
-            if not existing:
-                break
-            ids = [r["id"] for r in existing if r.get("id")]
-            if ids:
-                await _index_delete(tenant_id, kb_id, {"id": ids})
-            if len(existing) < 1000:
-                break
-            offset += 1000
+        # Delete by condition in one operation. Paging through matches while
+        # deleting them shifts subsequent offsets and leaves every later page
+        # behind, which is especially visible on large timeline graphs.
+        await _index_delete(tenant_id, kb_id, del_cond)
         # Also delete the meta rows
         meta_id = _meta_row_id(kb_id, compile_kwd, template_id)
         try:
@@ -728,9 +714,7 @@ async def _do_build(
     if incremental and last_build_time is not None:
         cleanup_ok = await _cleanup_deleted_docs(tenant_id, kb_id, compile_kwd, template_id)
 
-    # ── Scan doc_graph rows ──────────────────────────────────────────
-    # Backward compat: existing doc_graph rows don't have scope_kwd.
-    # Search for (scope_kwd="doc" OR (not exists scope_kwd)) AND compile_kwd.
+    # ── Scan document-level graph rows ───────────────────────────────
     base_cond = {
         "compile_kwd": [compile_kwd],
         "knowledge_graph_kwd": ["entity", "relation"],
@@ -738,8 +722,10 @@ async def _do_build(
     if template_id:
         base_cond["compilation_template_ids"] = [template_id]
     scan_all = not incremental or last_build_time is None
-    if not scan_all:
-        base_cond["scope_kwd"] = [_SCOPE_KWD_DOC]
+    # A full rebuild must only scan document-level rows.  Dataset-level rows
+    # are the output of this task and must never become input on the next run.
+    # This also makes disabled-document filtering effective after a rebuild.
+    base_cond["scope_kwd"] = [_SCOPE_KWD_DOC]
 
     # Build timestamp filter for incremental mode
     ts_cond = None