Merge synced face assignments against what is held, read once per pass

After 9cff677 the loop over the other device's confirmed and ignored faces
(13,000 on the reference library) still asked three cached statements per
face -- the person by uuid, the face's current assignment, and whether this
pair was rejected here. It was 51 ms of a steady-state merge.

The person is now resolved in the statement that reads the incoming rows,
by the local `people.uuid` key:

  SCAN fp
  SEARCH p USING INTEGER PRIMARY KEY (rowid=?)
  SEARCH lp USING COVERING INDEX sqlite_autoindex_people_1 (uuid=?)

and the local `face_person` (16,800 rows) and `face_person_rejected` are
each read once into memory and looked up there. A write goes to the table
and to the map, so a second remote face matched to the same local face sees
what the first left, as it did when each face re-read the table. The
incoming rows are ordered by face id -- the order the table was already
walked in -- since which of two such faces is applied last decides the
answer. An inner join to `people` drops the rows the old loop skipped for
want of a local person, and the counts in the report are unchanged.

After: the loop 10-12 ms. The merge as a whole, with the two changes before
this, went from 228-231 ms to 135 ms best of 5, and every catalog table
checksums the same after the bench as after the old build's run.

core/dr-catalog/src/merge.rs

@@ -927,29 +927,37 @@ fn merge_people_within(tx: &Connection, report: &mut MergeReport) -> Result<(),
 
     // ---- confirmations, and the anchors under an ignored group -----------
     {
+        // The person is resolved in the same statement, by the local
+        // `people.uuid` key, and what this device already holds is read once
+        // for the whole pass and looked up in memory. A steady-state pass
+        // walks every confirmed and every ignored face the other device
+        // holds -- 13,000 on the reference library -- and three statements
+        // per face, even cached, were 52 ms of it. In the key's order, which
+        // is the order the table is walked in anyway: when two of its faces
+        // match one of ours, which one is applied last decides the answer.
         let mut stmt = tx.prepare(&format!(
-            "SELECT fp.face_id, p.uuid, fp.probability, fp.confirmed
+            "SELECT fp.face_id, lp.id, fp.probability, fp.confirmed
                    FROM remote_cat.face_person fp
                    JOIN remote_cat.people p ON p.id = fp.person_id
-                  WHERE fp.confirmed = 1 OR {} = 1",
+                   JOIN main.people lp ON lp.uuid = p.uuid
+                  WHERE fp.confirmed = 1 OR {} = 1
+                  ORDER BY fp.face_id",
             if ignored_col { "p.ignored" } else { "0" }
         ))?;
-        let incoming: Vec<(i64, String, f64, bool)> = stmt
+        let incoming: Vec<(i64, i64, f64, bool)> = stmt
             .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))?
             .collect::<Result<_, _>>()?;
 
-        // Cached, not prepared per row: a steady-state pass walks every
-        // confirmed and every ignored face the other device holds -- 13,000 on
-        // the reference library -- and preparing four statements for each was
-        // most of the 780 ms a merge that changed nothing cost.
-        let mut person_of = tx.prepare_cached("SELECT id FROM people WHERE uuid = ?1")?;
-        let mut current = tx.prepare_cached(
-            "SELECT person_id, probability, confirmed FROM face_person WHERE face_id = ?1",
-        )?;
-        let mut rejected_q = tx.prepare_cached(
-            "SELECT EXISTS(SELECT 1 FROM face_person_rejected
-                            WHERE face_id = ?1 AND person_id = ?2)",
-        )?;
+        // Kept current as the loop writes: two of the other device's faces
+        // can match one of ours, and the second must see what the first left.
+        let mut held: std::collections::HashMap<i64, (i64, f64, i64)> = tx
+            .prepare("SELECT face_id, person_id, probability, confirmed FROM main.face_person")?
+            .query_map([], |r| Ok((r.get(0)?, (r.get(1)?, r.get(2)?, r.get(3)?))))?
+            .collect::<Result<_, _>>()?;
+        let rejected: std::collections::HashSet<(i64, i64)> = tx
+            .prepare("SELECT face_id, person_id FROM main.face_person_rejected")?
+            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
+            .collect::<Result<_, _>>()?;
         let mut assign = tx.prepare_cached(
             "INSERT INTO face_person (face_id, person_id, probability, confirmed)
              VALUES (?1, ?2, ?3, ?4)
@@ -959,15 +967,11 @@ fn merge_people_within(tx: &Connection, report: &mut MergeReport) -> Result<(),
                  confirmed = excluded.confirmed",
         )?;
 
-        for (remote_face, uuid, probability, confirmed) in incoming {
+        for (remote_face, person, probability, confirmed) in incoming {
             let Some(&local_face) = face_map.get(&remote_face) else {
                 continue;
             };
-            let person: Option<i64> = person_of.query_row([&uuid], |r| r.get(0)).optional()?;
-            let Some(person) = person else { continue };
-            let held: Option<(i64, f64, i64)> = current
-                .query_row([local_face], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
-                .optional()?;
+            let current = held.get(&local_face).copied();
 
             // A local confirmation is never overwritten, in either direction.
             // Two devices confirming the same face as different people is a
@@ -975,7 +979,7 @@ fn merge_people_within(tx: &Connection, report: &mut MergeReport) -> Result<(),
             // settle it with; silently taking the remote's answer would let a
             // sync undo something the user did here. It stays as it is, and the
             // user can change it on the device they are looking at.
-            if held.is_some_and(|(_, _, confirmed)| confirmed == 1) {
+            if current.is_some_and(|(_, _, confirmed)| confirmed == 1) {
                 report.faces_kept_local += 1;
                 continue;
             }
@@ -984,21 +988,22 @@ fn merge_people_within(tx: &Connection, report: &mut MergeReport) -> Result<(),
             // this user's judgement about this pair, and re-suggesting what
             // they pushed away is the behaviour that makes the feature feel
             // broken.
-            let rejected: bool = rejected_q.query_row([local_face, person], |r| r.get(0))?;
-            if rejected {
+            if rejected.contains(&(local_face, person)) {
                 continue;
             }
 
             // Written only when it differs. Rewriting a row with the values it
             // already holds dirtied a page per face, every pass, for nothing;
             // the report still counts it, as it always has.
-            if held != Some((person, probability, i64::from(confirmed))) {
+            let wanted = (person, probability, i64::from(confirmed));
+            if current != Some(wanted) {
                 assign.execute(rusqlite::params![
                     local_face,
                     person,
                     probability,
                     confirmed
                 ])?;
+                held.insert(local_face, wanted);
             }
             report.faces_assigned += 1;
         }