sqlite-utils optimize command, .optimize() and .detect_fts() table methods

docs/cli.rst

@@ -27,3 +27,18 @@ Vacuum
 You can run VACUUM to optimize your database like so::
 
     $ sqlite-utils vacuum mydb.db
+
+Optimize
+========
+
+The optimize command can dramatically reduce the size of your database if you are using SQLite full-text search. It runs OPTIMIZE against all of our FTS4 and FTS5 tables, then runs VACUUM.
+
+If you just want to run OPTIMIZE without the VACUUM, use the ``--no-vacuum`` flag.
+
+::
+
+    # Optimize all FTS tables and then VACUUM
+    $ sqlite-utils optimize mydb.db
+
+    # Optimize but skip the VACUUM
+    $ sqlite-utils optimize --no-vacuum mydb.db

docs/python-api.rst

@@ -298,6 +298,25 @@ If you insert additional records into the table you will need to refresh the sea
     }, pk="id")
     dogs.populate_fts(["name", "twitter"])
 
+``.enable_fts()`` defaults to using `FTS5 <https://www.sqlite.org/fts5.html>`__. If you wish to use `FTS4 <https://www.sqlite.org/fts3.html>`__ instead, use the following:
+
+.. code-block:: python
+
+    dogs.enable_fts(["name", "twitter"], fts_version="FTS4")
+
+Optimizing a full-text search table
+===================================
+
+Once you have populated a FTS table you can optimize it to dramatically reduce its size like so:
+
+.. code-block:: python
+
+    dogs.optimize()
+
+This runs the following SQL::
+
+    INSERT INTO dogs_fts (dogs_fts) VALUES ("optimize");
+
 Creating indexes
 ================
 

sqlite_utils/cli.py

@@ -37,3 +37,21 @@ def table_names(path, fts4, fts5):
 def vacuum(path):
     """Run VACUUM against the database"""
     sqlite_utils.Database(path).vacuum()
+
+
+@cli.command()
+@click.argument(
+    "path",
+    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
+    required=True,
+)
+@click.option("--no-vacuum", help="Don't run VACUUM", default=False, is_flag=True)
+def optimize(path, no_vacuum):
+    """Optimize all FTS tables and then run VACUUM - should shrink the database file"""
+    db = sqlite_utils.Database(path)
+    tables = db.table_names(fts4=True) + db.table_names(fts5=True)
+    with db.conn:
+        for table in tables:
+            db[table].optimize()
+    if not no_vacuum:
+        db.vacuum()

sqlite_utils/db.py

@@ -243,6 +243,40 @@ class Table:
         self.db.conn.executescript(sql)
         return self
 
+    def detect_fts(self):
+        "Detect if table has a corresponding FTS virtual table and return it"
+        rows = self.db.conn.execute(
+            """
+            SELECT name FROM sqlite_master
+                WHERE rootpage = 0
+                AND (
+                    sql LIKE '%VIRTUAL TABLE%USING FTS%content="{table}"%'
+                    OR (
+                        tbl_name = "{table}"
+                        AND sql LIKE '%VIRTUAL TABLE%USING FTS%'
+                    )
+                )
+        """.format(
+                table=self.name
+            )
+        ).fetchall()
+        if len(rows) == 0:
+            return None
+        else:
+            return rows[0][0]
+
+    def optimize(self):
+        fts_table = self.detect_fts()
+        if fts_table is not None:
+            self.db.conn.execute(
+                """
+                INSERT INTO [{table}] ([{table}]) VALUES ("optimize");
+            """.format(
+                    table=fts_table
+                )
+            )
+        return self
+
     def detect_column_types(self, records):
         all_column_types = {}
         for record in records:

tests/test_cli.py

@@ -1,5 +1,6 @@
 from sqlite_utils import cli, Database
 from click.testing import CliRunner
+import os
 import pytest
 import sqlite3
 
@@ -38,3 +39,26 @@ def test_table_names_fts5(db_path):
 def test_vacuum(db_path):
     result = CliRunner().invoke(cli.cli, ["vacuum", db_path])
     assert 0 == result.exit_code
+
+
+def test_optimize(db_path):
+    db = Database(db_path)
+    with db.conn:
+        for table in ("Gosh", "Gosh2"):
+            db[table].insert_all(
+                [
+                    {
+                        "c1": "verb{}".format(i),
+                        "c2": "noun{}".format(i),
+                        "c3": "adjective{}".format(i),
+                    }
+                    for i in range(10000)
+                ]
+            )
+        db["Gosh"].enable_fts(["c1", "c2", "c3"], fts_version="FTS4")
+        db["Gosh2"].enable_fts(["c1", "c2", "c3"], fts_version="FTS5")
+    size_before_optimize = os.stat(db_path).st_size
+    result = CliRunner().invoke(cli.cli, ["optimize", db_path])
+    assert 0 == result.exit_code
+    size_after_optimize = os.stat(db_path).st_size
+    assert size_after_optimize < size_before_optimize

tests/test_enable_fts.py

@@ -32,3 +32,19 @@ def test_populate_fts(fresh_db):
     # Now run populate_fts to make this record available
     table.populate_fts(["text", "country"])
     assert [("racoons are trash pandas", "USA", "bar")] == table.search("usa")
+
+
+def test_optimize_fts(fresh_db):
+    for fts_version in ("4", "5"):
+        table_name = "searchable_{}".format(fts_version)
+        table = fresh_db[table_name]
+        table.insert_all(search_records)
+        table.enable_fts(["text", "country"], fts_version="FTS{}".format(fts_version))
+    # You can call optimize successfully against the tables OR their _fts equivalents:
+    for table_name in (
+        "searchable_4",
+        "searchable_5",
+        "searchable_4_fts",
+        "searchable_5_fts",
+    ):
+        fresh_db[table_name].optimize()

tests/test_introspect.py

@@ -6,14 +6,30 @@ def test_table_names(existing_db):
 
 
 def test_table_names_fts4(existing_db):
-    existing_db["woo"].insert({"title": "Hello"})
-    existing_db["woo2"].insert({"title": "Hello"})
-    existing_db["woo"].enable_fts(["title"], fts_version="FTS4")
-    existing_db["woo2"].enable_fts(["title"], fts_version="FTS5")
+    existing_db["woo"].insert({"title": "Hello"}).enable_fts(
+        ["title"], fts_version="FTS4"
+    )
+    existing_db["woo2"].insert({"title": "Hello"}).enable_fts(
+        ["title"], fts_version="FTS5"
+    )
     assert ["woo_fts"] == existing_db.table_names(fts4=True)
     assert ["woo2_fts"] == existing_db.table_names(fts5=True)
 
 
+def test_detect_fts(existing_db):
+    existing_db["woo"].insert({"title": "Hello"}).enable_fts(
+        ["title"], fts_version="FTS4"
+    )
+    existing_db["woo2"].insert({"title": "Hello"}).enable_fts(
+        ["title"], fts_version="FTS5"
+    )
+    assert "woo_fts" == existing_db["woo"].detect_fts()
+    assert "woo_fts" == existing_db["woo_fts"].detect_fts()
+    assert "woo2_fts" == existing_db["woo2"].detect_fts()
+    assert "woo2_fts" == existing_db["woo2_fts"].detect_fts()
+    assert None == existing_db["foo"].detect_fts()
+
+
 def test_tables(existing_db):
     assert 1 == len(existing_db.tables)
     assert "foo" == existing_db.tables[0].name