|
19 | 19 |
|
20 | 20 | # Optional database-specific imports |
21 | 21 | try: |
22 | | - from sqlalchemy import text |
| 22 | + from sqlalchemy import text, bindparam |
23 | 23 | except ImportError: |
24 | 24 | text = None |
25 | 25 |
|
@@ -848,17 +848,49 @@ def cleanup_orphaned_collections( |
848 | 848 | # CRITICAL: Clean up orphaned chunks within active KB collections |
849 | 849 | # KB collections may contain chunks referencing deleted files |
850 | 850 | # This handles the case where a file is deleted but the KB collection remains active |
| 851 | + # NOTE: We use the active_file_ids set instead of querying the `file` table directly, |
| 852 | + # because in split-DB deployments the `file` table lives in a separate database |
| 853 | + # and cannot be referenced from the vector database. |
851 | 854 | orphaned_chunks_deleted = 0 |
852 | 855 | try: |
853 | | - if self.session: |
| 856 | + if self.session and active_file_ids: |
854 | 857 | log.debug("Cleaning orphaned chunks from active KB collections") |
| 858 | + # First, find all distinct file_ids referenced by chunks |
| 859 | + file_id_result = self.session.execute(text(""" |
| 860 | + SELECT DISTINCT dc.vmetadata->>'file_id' AS file_id |
| 861 | + FROM document_chunk dc |
| 862 | + WHERE dc.vmetadata ? 'file_id' |
| 863 | + AND dc.vmetadata->>'file_id' IS NOT NULL |
| 864 | + """)) |
| 865 | + referenced_file_ids = {row[0] for row in file_id_result} |
| 866 | + |
| 867 | + # Determine which referenced file_ids are orphaned (not in active set) |
| 868 | + orphaned_file_ids = referenced_file_ids - active_file_ids |
| 869 | + if orphaned_file_ids: |
| 870 | + # Delete chunks referencing orphaned file_ids in batches |
| 871 | + orphaned_list = list(orphaned_file_ids) |
| 872 | + batch_size = 500 |
| 873 | + for i in range(0, len(orphaned_list), batch_size): |
| 874 | + batch = orphaned_list[i:i + batch_size] |
| 875 | + result = self.session.execute( |
| 876 | + text(""" |
| 877 | + DELETE FROM document_chunk dc |
| 878 | + WHERE dc.vmetadata ? 'file_id' |
| 879 | + AND dc.vmetadata->>'file_id' IN :file_ids |
| 880 | + """).bindparams(bindparam('file_ids', expanding=True)), |
| 881 | + {'file_ids': batch} |
| 882 | + ) |
| 883 | + orphaned_chunks_deleted += result.rowcount |
| 884 | + self.session.commit() |
| 885 | + if orphaned_chunks_deleted > 0: |
| 886 | + log.info(f"Deleted {orphaned_chunks_deleted} orphaned chunks from active collections") |
| 887 | + elif self.session: |
| 888 | + log.debug("Cleaning orphaned chunks from active KB collections (no active files)") |
| 889 | + # If there are no active file IDs, all chunks with file_id metadata are orphaned |
855 | 890 | result = self.session.execute(text(""" |
856 | 891 | DELETE FROM document_chunk dc |
857 | 892 | WHERE dc.vmetadata ? 'file_id' |
858 | | - AND NOT EXISTS ( |
859 | | - SELECT 1 FROM file f |
860 | | - WHERE f.id = (dc.vmetadata->>'file_id') |
861 | | - ) |
| 893 | + AND dc.vmetadata->>'file_id' IS NOT NULL |
862 | 894 | """)) |
863 | 895 | orphaned_chunks_deleted = result.rowcount |
864 | 896 | self.session.commit() |
|
0 commit comments