Fix FTS4 triggers for updates and deletes

sqlite_utils/db.py

@@ -3794,10 +3794,18 @@ def enable_fts(
             fts_schema = self.db[f"{self.name}_fts"].schema
             if fts_schema != create_fts_sql:
                 should_recreate = True
-            expected_triggers = {self.name + suffix for suffix in ("_ai", "_ad", "_au")}
+            trigger_suffixes = ["_ai", "_ad", "_au"]
+            if fts_version.upper() == "FTS4":
+                trigger_suffixes.append("_bu")
+            expected_triggers = {self.name + suffix for suffix in trigger_suffixes}
+            known_triggers = {
+                self.name + suffix for suffix in ("_ai", "_ad", "_au", "_bu")
+            }
             existing_triggers = {t.name for t in self.triggers}
             has_triggers = existing_triggers.issuperset(expected_triggers)
-            if has_triggers != create_triggers:
+            if (create_triggers and not has_triggers) or (
+                not create_triggers and bool(existing_triggers & known_triggers)
+            ):
                 should_recreate = True
             if not should_recreate:
                 # Table with correct configuration already exists
@@ -3815,31 +3823,62 @@ def enable_fts(
             columns_quoted = ", ".join(quote_identifier(c) for c in columns)
             table = quote_identifier(self.name)
             table_fts = quote_identifier(self.name + "_fts")
-            triggers = (
-                textwrap.dedent("""
-                CREATE TRIGGER {table_ai} AFTER INSERT ON {table} BEGIN
-                  INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
-                END;
-                CREATE TRIGGER {table_ad} AFTER DELETE ON {table} BEGIN
-                  INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
-                END;
-                CREATE TRIGGER {table_au} AFTER UPDATE ON {table} BEGIN
-                  INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
-                  INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
-                END;
-            """)
-                .strip()
-                .format(
-                    table=table,
-                    table_fts=table_fts,
-                    table_ai=quote_identifier(self.name + "_ai"),
-                    table_ad=quote_identifier(self.name + "_ad"),
-                    table_au=quote_identifier(self.name + "_au"),
-                    columns=columns_quoted,
-                    old_cols=old_cols,
-                    new_cols=new_cols,
+            if fts_version.upper() == "FTS4":
+                # FTS4 reads old values from the content table while deleting
+                # index entries, so this must run before the content changes.
+                triggers = (
+                    textwrap.dedent("""
+                    CREATE TRIGGER {table_ai} AFTER INSERT ON {table} BEGIN
+                      INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
+                    END;
+                    CREATE TRIGGER {table_ad} BEFORE DELETE ON {table} BEGIN
+                      DELETE FROM {table_fts} WHERE rowid = old.rowid;
+                    END;
+                    CREATE TRIGGER {table_bu} BEFORE UPDATE ON {table} BEGIN
+                      DELETE FROM {table_fts} WHERE rowid = old.rowid;
+                    END;
+                    CREATE TRIGGER {table_au} AFTER UPDATE ON {table} BEGIN
+                      INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
+                    END;
+                """)
+                    .strip()
+                    .format(
+                        table=table,
+                        table_fts=table_fts,
+                        table_ai=quote_identifier(self.name + "_ai"),
+                        table_ad=quote_identifier(self.name + "_ad"),
+                        table_bu=quote_identifier(self.name + "_bu"),
+                        table_au=quote_identifier(self.name + "_au"),
+                        columns=columns_quoted,
+                        new_cols=new_cols,
+                    )
+                )
+            else:
+                triggers = (
+                    textwrap.dedent("""
+                    CREATE TRIGGER {table_ai} AFTER INSERT ON {table} BEGIN
+                      INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
+                    END;
+                    CREATE TRIGGER {table_ad} AFTER DELETE ON {table} BEGIN
+                      INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
+                    END;
+                    CREATE TRIGGER {table_au} AFTER UPDATE ON {table} BEGIN
+                      INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
+                      INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
+                    END;
+                """)
+                    .strip()
+                    .format(
+                        table=table,
+                        table_fts=table_fts,
+                        table_ai=quote_identifier(self.name + "_ai"),
+                        table_ad=quote_identifier(self.name + "_ad"),
+                        table_au=quote_identifier(self.name + "_au"),
+                        columns=columns_quoted,
+                        old_cols=old_cols,
+                        new_cols=new_cols,
+                    )
                 )
-            )
             self.db.executescript(triggers)
         return self
 
@@ -3869,16 +3908,30 @@ def populate_fts(self, columns: Iterable[str]) -> "Table":
     def disable_fts(self) -> "Table":
         "Remove any full-text search index and related triggers configured for this table."
         fts_table = self.detect_fts()
-        if fts_table:
-            self.db[fts_table].drop()
+        if fts_table is None:
+            return self
+        self.db[fts_table].drop()
         # Now delete the triggers that related to that table
         sql = textwrap.dedent("""
             SELECT name FROM sqlite_master
                 WHERE type = 'trigger'
-                AND (sql LIKE '% INSERT INTO [{}]%' OR sql LIKE '% INSERT INTO "{}"%')
-        """).strip().format(fts_table, fts_table)
+                AND (
+                    instr(lower(sql), lower(:insert_quoted)) > 0
+                    OR instr(lower(sql), lower(:insert_bracket)) > 0
+                    OR instr(lower(sql), lower(:delete_quoted)) > 0
+                    OR instr(lower(sql), lower(:delete_bracket)) > 0
+                )
+        """).strip()
+        quoted = quote_identifier(fts_table)
+        bracketed = f"[{fts_table}]"
+        patterns = {
+            "insert_quoted": f" INSERT INTO {quoted}",
+            "insert_bracket": f" INSERT INTO {bracketed}",
+            "delete_quoted": f" DELETE FROM {quoted}",
+            "delete_bracket": f" DELETE FROM {bracketed}",
+        }
         trigger_names = []
-        for row in self.db.execute(sql).fetchall():
+        for row in self.db.execute(sql, patterns).fetchall():
             trigger_names.append(row[0])
         with self.db.atomic():
             for trigger_name in trigger_names:

tests/test_fts.py

@@ -319,6 +319,129 @@ def test_enable_fts_with_triggers(fresh_db):
     assert [] == list(table.search("bar"))
 
 
+@pytest.mark.parametrize("fts_version", ["FTS4", "FTS5"])
+def test_fts_triggers_keep_updates_and_deletes_in_sync(fresh_db, fts_version):
+    table = fresh_db.table("notes '100%'")
+    table.insert({"id": 1, "text": "alpha", "country": "Japan"}, pk="id")
+    table.enable_fts(["text", "country"], fts_version=fts_version, create_triggers=True)
+
+    table.insert({"id": 2, "text": "otter", "country": "Canada"})
+    assert [row["id"] for row in table.search("otter")] == [2]
+    table.update(1, {"text": "beta", "country": "Iceland"})
+    assert list(table.search("alpha")) == []
+    assert [row["id"] for row in table.search("beta")] == [1]
+    assert list(table.search("Japan")) == []
+    assert [row["id"] for row in table.search("Iceland")] == [1]
+    table.delete(2)
+    assert list(table.search("otter")) == []
+
+    table.disable_fts()
+    assert not table.triggers
+    table.update(1, {"text": "gamma"})
+    table.delete(1)
+
+
+def test_fts4_replace_with_same_configuration_preserves_index(fresh_db):
+    table = fresh_db.table("notes")
+    table.insert({"id": 1, "text": "alpha"}, pk="id")
+    table.enable_fts(["text"], fts_version="FTS4", create_triggers=True)
+    triggers = {trigger.name for trigger in table.triggers}
+    assert triggers == {"notes_ai", "notes_ad", "notes_bu", "notes_au"}
+
+    table.enable_fts(["text"], fts_version="FTS4", create_triggers=True, replace=True)
+    assert {trigger.name for trigger in table.triggers} == triggers
+    table.update(1, {"text": "beta"})
+    assert list(table.search("alpha")) == []
+    assert [row["id"] for row in table.search("beta")] == [1]
+
+
+def test_fts4_trigger_tracks_changed_rowid(fresh_db):
+    table = fresh_db.table("notes")
+    table.insert({"id": 1, "text": "alpha"}, pk="id")
+    table.enable_fts(["text"], fts_version="FTS4", create_triggers=True)
+
+    table.update(1, {"id": 2, "text": "beta"})
+    assert list(table.search("alpha")) == []
+    assert [row["id"] for row in table.search("beta")] == [2]
+    table.delete(2)
+    assert list(table.search("beta")) == []
+
+
+def test_replace_switches_fts_versions_with_triggers(fresh_db):
+    table = fresh_db.table("notes")
+    table.insert({"id": 1, "text": "alpha"}, pk="id")
+    table.enable_fts(["text"], fts_version="FTS4", create_triggers=True)
+
+    table.enable_fts(["text"], fts_version="FTS5", create_triggers=True, replace=True)
+    assert {trigger.name for trigger in table.triggers} == {
+        "notes_ai",
+        "notes_ad",
+        "notes_au",
+    }
+    table.update(1, {"text": "beta"})
+    assert list(table.search("alpha")) == []
+    assert [row["id"] for row in table.search("beta")] == [1]
+
+    table.enable_fts(["text"], fts_version="FTS4", create_triggers=True, replace=True)
+    assert {trigger.name for trigger in table.triggers} == {
+        "notes_ai",
+        "notes_ad",
+        "notes_bu",
+        "notes_au",
+    }
+    table.update(1, {"text": "gamma"})
+    assert list(table.search("beta")) == []
+    assert [row["id"] for row in table.search("gamma")] == [1]
+
+
+def test_disable_fts_keeps_triggers_for_similarly_named_table(fresh_db):
+    first = fresh_db.table("notes_%")
+    other = fresh_db.table("notes_123")
+    first.insert({"text": "first"})
+    other.insert({"text": "second"})
+    first.enable_fts(["text"], fts_version="FTS4", create_triggers=True)
+    other.enable_fts(["text"], fts_version="FTS4", create_triggers=True)
+    other_triggers = {trigger.name for trigger in other.triggers}
+
+    first.disable_fts()
+    assert not first.triggers
+    assert {trigger.name for trigger in other.triggers} == other_triggers
+    other.insert({"text": "third"})
+    assert len(list(other.search("third"))) == 1
+
+
+@pytest.mark.parametrize("create_triggers", [True, False])
+def test_fts4_replace_migrates_legacy_triggers(fresh_db, create_triggers):
+    table = fresh_db.table("notes")
+    table.insert({"id": 1, "text": "alpha"}, pk="id")
+    table.enable_fts(["text"], fts_version="FTS4")
+    # Recreate the three FTS5-style triggers generated by older versions.
+    fresh_db.executescript("""
+        CREATE TRIGGER notes_ai AFTER INSERT ON notes BEGIN
+          INSERT INTO "notes_fts" (rowid, text) VALUES (new.rowid, new.text);
+        END;
+        CREATE TRIGGER notes_ad AFTER DELETE ON notes BEGIN
+          INSERT INTO "notes_fts" (notes_fts, rowid, text) VALUES('delete', old.rowid, old.text);
+        END;
+        CREATE TRIGGER notes_au AFTER UPDATE ON notes BEGIN
+          INSERT INTO "notes_fts" (notes_fts, rowid, text) VALUES('delete', old.rowid, old.text);
+          INSERT INTO "notes_fts" (rowid, text) VALUES (new.rowid, new.text);
+        END;
+    """)
+
+    table.enable_fts(
+        ["text"], fts_version="FTS4", create_triggers=create_triggers, replace=True
+    )
+    expected = (
+        {"notes_ai", "notes_ad", "notes_bu", "notes_au"} if create_triggers else set()
+    )
+    assert {trigger.name for trigger in table.triggers} == expected
+    table.update(1, {"text": "beta"})
+    if create_triggers:
+        assert list(table.search("alpha")) == []
+        assert [row["id"] for row in table.search("beta")] == [1]
+
+
 @pytest.mark.parametrize("create_triggers", [True, False])
 def test_disable_fts(fresh_db, create_triggers):
     table = fresh_db.table("searchable")

tests/test_introspect.py

@@ -235,7 +235,8 @@ def test_checks_nonexistent_and_virtual_tables(fresh_db):
     assert fresh_db.table("searchable_fts").checks == []
 
 
-def test_triggers_and_triggers_dict(fresh_db):
+@pytest.mark.parametrize("fts_version", ["FTS4", "FTS5"])
+def test_triggers_and_triggers_dict(fresh_db, fts_version):
     assert [] == fresh_db.triggers
     authors = fresh_db.table("authors")
     authors.insert_all(
@@ -250,13 +251,15 @@ def test_triggers_and_triggers_dict(fresh_db):
     assert fresh_db.table("other").triggers == []
     assert fresh_db.triggers_dict == {}
     authors.enable_fts(
-        ["name", "famous_works"], fts_version="FTS4", create_triggers=True
+        ["name", "famous_works"], fts_version=fts_version, create_triggers=True
     )
     expected_triggers = {
         ("authors_ai", "authors"),
         ("authors_ad", "authors"),
         ("authors_au", "authors"),
     }
+    if fts_version == "FTS4":
+        expected_triggers.add(("authors_bu", "authors"))
     assert expected_triggers == {(t.name, t.table) for t in fresh_db.triggers}
     assert expected_triggers == {
         (t.name, t.table) for t in fresh_db.table("authors").triggers
@@ -278,6 +281,26 @@ def test_triggers_and_triggers_dict(fresh_db):
             '  INSERT INTO "authors_fts" (rowid, "name", "famous_works") VALUES (new.rowid, new."name", new."famous_works");\nEND'
         ),
     }
+    if fts_version == "FTS4":
+        expected_triggers.update(
+            {
+                "authors_ad": (
+                    'CREATE TRIGGER "authors_ad" BEFORE DELETE ON "authors" BEGIN\n'
+                    '  DELETE FROM "authors_fts" WHERE rowid = old.rowid;\n'
+                    "END"
+                ),
+                "authors_bu": (
+                    'CREATE TRIGGER "authors_bu" BEFORE UPDATE ON "authors" BEGIN\n'
+                    '  DELETE FROM "authors_fts" WHERE rowid = old.rowid;\n'
+                    "END"
+                ),
+                "authors_au": (
+                    'CREATE TRIGGER "authors_au" AFTER UPDATE ON "authors" BEGIN\n'
+                    '  INSERT INTO "authors_fts" (rowid, "name", "famous_works") VALUES (new.rowid, new."name", new."famous_works");\n'
+                    "END"
+                ),
+            }
+        )
     assert authors.triggers_dict == expected_triggers
     assert fresh_db.table("other").triggers == []
     assert fresh_db.table("other").triggers_dict == {}