Add method to re-index FTS

resources/fts_triggers.sql

@@ -1,17 +1,17 @@
-CREATE TEMP TRIGGER fts_db_book_deleted_trg AFTER DELETE ON main.books BEGIN
+CREATE TEMP TRIGGER IF NOT EXISTS fts_db_book_deleted_trg AFTER DELETE ON main.books BEGIN
     DELETE FROM books_text WHERE book=OLD.id;
     DELETE FROM dirtied_formats WHERE book=OLD.id;
 END;
 
-CREATE TEMP TRIGGER fts_db_format_deleted_trg AFTER DELETE ON main.data BEGIN
+CREATE TEMP TRIGGER IF NOT EXISTS fts_db_format_deleted_trg AFTER DELETE ON main.data BEGIN
     DELETE FROM books_text WHERE book=OLD.book AND format=OLD.format;
     DELETE FROM dirtied_formats WHERE book=OLD.book AND format=OLD.format;
 END;
 
-CREATE TEMP TRIGGER fts_db_format_added_trg AFTER INSERT ON main.data BEGIN
+CREATE TEMP TRIGGER IF NOT EXISTS fts_db_format_added_trg AFTER INSERT ON main.data BEGIN
     INSERT OR IGNORE INTO dirtied_formats(book, format) VALUES (NEW.book, NEW.format);
 END;
 
-CREATE TEMP TRIGGER fts_db_format_updated_trg AFTER UPDATE ON main.data BEGIN
+CREATE TEMP TRIGGER IF NOT EXISTS fts_db_format_updated_trg AFTER UPDATE ON main.data BEGIN
     INSERT OR IGNORE INTO dirtied_formats(book, format) VALUES (NEW.book, NEW.format);
 END;

src/calibre/db/backend.py

@@ -933,6 +933,7 @@ class DB:
             return
         from .fts.connect import FTS
         self.fts = FTS(dbref)
+        return self.fts
 
     def enable_fts(self, dbref=None):
         enabled = dbref is not None
@@ -962,6 +963,12 @@ class DB:
     def get_next_fts_job(self):
         return self.fts.get_next_fts_job()
 
+    def reindex_fts(self):
+        if self.conn.fts_dbpath:
+            self.conn.execute('DETACH fts_db')
+            os.remove(self.conn.fts_dbpath)
+            self.conn.fts_dbpath = None
+
     def remove_dirty_fts(self, book_id, fmt):
         return self.fts.remove_dirty(book_id, fmt)
 

src/calibre/db/cache.py

@@ -431,9 +431,10 @@ class Cache:
     def initialize_fts(self):
         self.fts_queue_thread = None
         self.fts_job_queue = Queue()
-        self.backend.initialize_fts(weakref.ref(self))
+        fts = self.backend.initialize_fts(weakref.ref(self))
         if self.is_fts_enabled():
             self.start_fts_pool()
+        return fts
 
     def start_fts_pool(self):
         from threading import Thread
@@ -531,6 +532,19 @@ class Cache:
     def commit_fts_result(self, book_id, fmt, fmt_size, fmt_hash, text, err_msg):
         return self.backend.commit_fts_result(book_id, fmt, fmt_size, fmt_hash, text, err_msg)
 
+    @api
+    def reindex_fts(self):
+        if not self.is_fts_enabled():
+            return
+        with self.write_lock:
+            self._shutdown_fts()
+        self._shutdown_fts(stage=2)
+        with self.write_lock:
+            self.backend.reindex_fts()
+            fts = self.initialize_fts()
+            self._queue_next_fts_job()
+        return fts
+
     @api
     def set_fts_num_of_workers(self, num=None):
         existing = self.backend.fts_num_of_workers
@@ -2381,6 +2395,19 @@ class Cache:
     def __del__(self):
         self.close()
 
+    def _shutdown_fts(self, stage=1):
+        if stage == 1:
+            self.backend.shutdown_fts()
+            if self.fts_queue_thread is not None:
+                self.fts_job_queue.put(None)
+            return
+        # the fts supervisor thread could be in the middle of committing a
+        # result to the db, so holding a lock here will cause a deadlock
+        if self.fts_queue_thread is not None:
+            self.fts_queue_thread.join()
+            self.fts_queue_thread = None
+        self.backend.join_fts()
+
     @api
     def close(self):
         with self.write_lock:
@@ -2389,9 +2416,7 @@ class Cache:
             self.close_called = True
             self.shutting_down = True
             self.event_dispatcher.close()
-            self.backend.shutdown_fts()
+            self._shutdown_fts()
-            if self.fts_queue_thread is not None:
-                self.fts_job_queue.put(None)
             from calibre.customize.ui import available_library_closed_plugins
             for plugin in available_library_closed_plugins():
                 try:
@@ -2399,12 +2424,7 @@ class Cache:
                 except Exception:
                     import traceback
                     traceback.print_exc()
-        # the fts supervisor thread could be in the middle of committing a
+        self._shutdown_fts(stage=2)
-        # result to the db, so holding a lock here will cause a deadlock
-        if self.fts_queue_thread is not None:
-            self.fts_queue_thread.join()
-            self.fts_queue_thread = None
-        self.backend.join_fts()
         with self.write_lock:
             self.backend.close()
 

src/calibre/db/fts/pool.py

@@ -206,7 +206,8 @@ class Pool:
             self.initialized.clear()
 
     def join(self):
-        self.supervisor_thread.join()
+        with suppress(AttributeError):
+            self.supervisor_thread.join()
         for w in self.workers:
             w.join()
         self.workers = []

src/calibre/db/tests/fts_api.py

@@ -143,6 +143,12 @@ class FTSAPITest(BaseTest):
             'some other long text that will [also] help with the testing of search'})
         self.ae({x['text'] for x in cache.fts_search('also', highlight_start='[', highlight_end=']', snippet_size=3)}, {
             '…will [also] help…'})
+        fts = cache.reindex_fts()
+        self.wait_for_fts_to_finish(fts)
+        self.assertFalse(fts.all_currently_dirty())
+        self.ae({x['id'] for x in cache.fts_search('help')}, {1, 2})
+        cache.remove_books((1,))
+        self.ae({x['id'] for x in cache.fts_search('help')}, {2})
         cache.close()
 
     def test_fts_triggers(self):