Part 6: Vacuum the fulltext_values table after clearing excisions.

Author: ncalexan

db/src/db.rs

@@ -3282,4 +3282,119 @@ mod tests {
                [?e :db.excise/beforeT {} ?tx4 true]
                [?tx4 :db/txInstant ?ms4 ?tx4 true]]]"#, report.tx_id));
     }
+
+    #[test]
+    fn test_excision_fulltext() {
+        let mut conn = TestConn::default();
+
+        assert_transact!(conn, r#"[
+            {:db/id 200
+             :db/ident :test/fulltext
+             :db/valueType :db.type/string
+             :db/cardinality :db.cardinality/one
+             :db/fulltext true
+             :db/index true}
+        ]"#);
+
+        assert_transact!(conn, r#"[
+            {:db/id 300
+             :test/fulltext "test1"}
+            {:db/id 301
+             :test/fulltext "test2"}
+        ]"#);
+
+        assert_transact!(conn, r#"[
+            {:db/id 300
+             :test/fulltext "test3"}
+            {:db/id 301
+             :test/fulltext "test4"}
+        ]"#);
+
+        // Before.
+        assert_matches!(conn.fulltext_values(), r#"
+           [[1 "test1"]
+            [2 "test2"]
+            [3 "test3"]
+            [4 "test4"]]"#);
+        assert_matches!(conn.datoms(), r#"
+            [[200 :db/ident :test/fulltext]
+             [200 :db/valueType :db.type/string]
+             [200 :db/cardinality :db.cardinality/one]
+             [200 :db/index true]
+             [200 :db/fulltext true]
+             [300 :test/fulltext 3]
+             [301 :test/fulltext 4]]"#);
+
+        let tempid_report = assert_transact!(conn, r#"[
+            [:db/add "e" :db/excise 300]
+        ]"#);
+        // This is implementation specific, but it should be deterministic.
+        assert_matches!(tempids(&tempid_report),
+                        "{\"e\" 65536}");
+
+        // After.
+        assert_matches!(conn.datoms(), r#"
+            [[200 :db/ident :test/fulltext]
+             [200 :db/valueType :db.type/string]
+             [200 :db/cardinality :db.cardinality/one]
+             [200 :db/index true]
+             [200 :db/fulltext true]
+             [301 :test/fulltext 4]
+             [?e :db/excise 300]]"#);
+
+        // We have enqueued a pending excision.
+        let pending = excision::pending_excisions(&conn.sqlite, &conn.partition_map, &conn.schema).expect("pending_excisions");
+        assert_eq!(pending, ::std::iter::once((65536, excision::Excision {
+            target: 300,
+            attrs: None,
+            before_tx: None,
+        })).collect());
+
+        // Before processing the pending excision, we have full transactions in the transaction log.
+        assert_matches!(conn.transactions(), r#"
+            [[[200 :db/ident :test/fulltext ?tx1 true]
+              [200 :db/valueType :db.type/string ?tx1 true]
+              [200 :db/cardinality :db.cardinality/one ?tx1 true]
+              [200 :db/index true ?tx1 true]
+              [200 :db/fulltext true ?tx1 true]
+              [?tx1 :db/txInstant ?ms ?tx1 true]]
+             [[300 :test/fulltext 1 ?tx2 true]
+              [301 :test/fulltext 2 ?tx2 true]
+              [?tx2 :db/txInstant ?ms2 ?tx2 true]]
+             [[300 :test/fulltext 1 ?tx3 false]
+              [300 :test/fulltext 3 ?tx3 true]
+              [301 :test/fulltext 2 ?tx3 false]
+              [301 :test/fulltext 4 ?tx3 true]
+              [?tx3 :db/txInstant ?ms3 ?tx3 true]]
+             [[?e :db/excise 300 ?tx4 true]
+              [?tx4 :db/txInstant ?ms4 ?tx4 true]]]"#);
+
+        excision::ensure_no_pending_excisions(&conn.sqlite, &conn.partition_map, &conn.schema).expect("ensure_no_pending_excisions");
+
+        // After processing the pending excision, we have nothing left pending.
+        let pending = excision::pending_excisions(&conn.sqlite, &conn.partition_map, &conn.schema).expect("pending_excisions");
+        assert_eq!(pending, Default::default());
+
+        // After processing the pending excision, we have rewritten transactions in the transaction
+        // log to not refer to the targeted attributes of the target entity.
+        assert_matches!(conn.transactions(), r#"
+            [[[200 :db/ident :test/fulltext ?tx1 true]
+              [200 :db/valueType :db.type/string ?tx1 true]
+              [200 :db/cardinality :db.cardinality/one ?tx1 true]
+              [200 :db/index true ?tx1 true]
+              [200 :db/fulltext true ?tx1 true]
+              [?tx1 :db/txInstant ?ms ?tx1 true]]
+             [[301 :test/fulltext 2 ?tx2 true]
+              [?tx2 :db/txInstant ?ms2 ?tx2 true]]
+             [[301 :test/fulltext 2 ?tx3 false]
+              [301 :test/fulltext 4 ?tx3 true]
+              [?tx3 :db/txInstant ?ms3 ?tx3 true]]
+             [[?e :db/excise 300 ?tx4 true]
+              [?tx4 :db/txInstant ?ms4 ?tx4 true]]]"#);
+
+        // After processing the pending excision, we have vacuumed dangling fulltext values.
+        assert_matches!(conn.fulltext_values(), r#"
+            [[2 "test2"]
+             [4 "test4"]]"#);
+    }
 }

db/src/excision.rs

@@ -22,10 +22,12 @@ use mentat_core::{
     Entid,
     HasSchema,
     Schema,
+    TypedValue,
 };
 
 use db::{
     PartitionMapping,
+    TypedSQLValue,
     read_materialized_transaction_aev_trie,
 };
 
@@ -258,5 +260,38 @@ pub(crate) fn ensure_no_pending_excisions(conn: &rusqlite::Connection, partition
 
     conn.execute("UPDATE excisions SET status = 0", &[])?;
 
+    // TODO: only vacuum fulltext if an excision (likely) impacted fulltext values, since this is
+    // very expensive.  As always, correctness first, performance second.
+    vacuum_fulltext_table(conn)?;
+
     Ok(list.into_iter().map(|(entity, excision, _status)| (entity, excision)).collect())
 }
+
+
+/// Delete fulltext values that are no longer refered to in the `datoms` or `transactions` table.
+pub(crate) fn vacuum_fulltext_table(conn: &rusqlite::Connection) -> Result<()> {
+    let (true_value, true_value_type_tag) = TypedValue::Boolean(true).to_sql_value_pair();
+
+    // First, collect all `:db/fulltext true` attributes.  This is easier than extracting them from
+    // a `Schema` (no need to execute multiple insertions for large collections), but less flexible.
+    conn.execute(r#"CREATE TABLE temp.fulltext_as (a SMALLINT NOT NULL)"# , &[])?;
+    conn.execute(r#"INSERT INTO temp.fulltext_as (a)
+                    SELECT e FROM schema WHERE a = ? AND v = ? AND value_type_tag = ?"# ,
+                 &[&entids::DB_FULLTEXT, &true_value, &true_value_type_tag])?;
+
+    // Next, purge values that aren't referenced.  We're using that `:db/fulltext true` attributes
+    // always have `:db/index true`, so that we can use the `avet` index.
+    conn.execute(r#"DELETE FROM fulltext_values
+                    WHERE rowid NOT IN
+                      (SELECT v
+                       FROM datoms
+                       WHERE index_avet IS NOT 0 AND a IN temp.fulltext_as
+                       UNION ALL
+                       SELECT v
+                       FROM transactions
+                       WHERE a IN temp.fulltext_as)"#, &[])?;
+
+    conn.execute(r#"DROP TABLE temp.fulltext_as"# , &[])?;
+
+    Ok(())
+}