Merge 'core/incremental: Fix re-insertion of data with same key' from Glauber Costa

There is currently a bug found in our materialized view implementation
that happens when we delete a row, and then re-insert another row with
the same primary key.
Our insert code needs to detect updates and generate a DELETE + INSERT.
But in this case, after the initial DELETE, the fresh insert generates
another delete.
We ended up with the wrong response for aggregations (and I am pretty
sure even filter-only views would manifest the bug as well), where
groups that should still be present just disappeared because of the
extra delete.
A new test case is added that fails without the fix.

Closes #3601

core/vdbe/execute.rs

@@ -5695,31 +5695,52 @@ pub fn op_insert(
 
                 turso_assert!(!flag.has(InsertFlags::REQUIRE_SEEK), "to capture old record accurately, we must be located at the correct position in the table");
 
+                // Get the key we're going to insert
+                let insert_key = match &state.registers[*key_reg].get_value() {
+                    Value::Integer(i) => *i,
+                    _ => {
+                        // If key is not an integer, we can't check - assume no old record
+                        state.op_insert_state.old_record = None;
+                        state.op_insert_state.sub_state = if flag.has(InsertFlags::REQUIRE_SEEK) {
+                            OpInsertSubState::Seek
+                        } else {
+                            OpInsertSubState::Insert
+                        };
+                        continue;
+                    }
+                };
+
                 let old_record = {
                     let cursor = state.get_cursor(*cursor_id);
                     let cursor = cursor.as_btree_mut();
                     // Get the current key - for INSERT operations, there may not be a current row
                     let maybe_key = return_if_io!(cursor.rowid());
                     if let Some(key) = maybe_key {
-                        // Get the current record before deletion and extract values
-                        let maybe_record = return_if_io!(cursor.record());
-                        if let Some(record) = maybe_record {
-                            let mut values = record
-                                .get_values()
-                                .into_iter()
-                                .map(|v| v.to_owned())
-                                .collect::<Vec<_>>();
+                        // Only capture as old record if the cursor is at the position we're inserting to
+                        if key == insert_key {
+                            // Get the current record before deletion and extract values
+                            let maybe_record = return_if_io!(cursor.record());
+                            if let Some(record) = maybe_record {
+                                let mut values = record
+                                    .get_values()
+                                    .into_iter()
+                                    .map(|v| v.to_owned())
+                                    .collect::<Vec<_>>();
 
-                            // Fix rowid alias columns: replace Null with actual rowid value
-                            if let Some(table) = schema.get_table(table_name) {
-                                for (i, col) in table.columns().iter().enumerate() {
-                                    if col.is_rowid_alias && i < values.len() {
-                                        values[i] = Value::Integer(key);
+                                // Fix rowid alias columns: replace Null with actual rowid value
+                                if let Some(table) = schema.get_table(table_name) {
+                                    for (i, col) in table.columns().iter().enumerate() {
+                                        if col.is_rowid_alias && i < values.len() {
+                                            values[i] = Value::Integer(key);
+                                        }
                                     }
                                 }
+                                Some((key, values))
+                            } else {
+                                None
                             }
-                            Some((key, values))
                         } else {
+                            // Cursor is at wrong position - this is a fresh INSERT, not a replacement
                             None
                         }
                     } else {

testing/materialized_views.test

@@ -3,6 +3,35 @@
 set testdir [file dirname $argv0]
 source $testdir/tester.tcl
 
+# Test that INSERT with reused primary keys maintains correct aggregate counts
+# When a row is deleted and a new row is inserted with the same primary key
+# but different group value, all groups should maintain correct counts
+do_execsql_test_on_specific_db {:memory:} matview-insert-reused-key-maintains-all-groups {
+    CREATE TABLE t(id INTEGER PRIMARY KEY, val TEXT);
+    INSERT INTO t VALUES (1, 'A'), (2, 'B');
+
+    CREATE MATERIALIZED VIEW v AS
+        SELECT val, COUNT(*) as cnt
+        FROM t
+        GROUP BY val;
+
+    -- Initial state: A=1, B=1
+    SELECT * FROM v ORDER BY val;
+
+    -- Delete id=1 (which has 'A')
+    DELETE FROM t WHERE id = 1;
+    SELECT * FROM v ORDER BY val;
+
+    -- Insert id=1 with different value 'C'
+    -- This should NOT affect group 'B'
+    INSERT INTO t VALUES (1, 'C');
+    SELECT * FROM v ORDER BY val;
+} {A|1
+B|1
+B|1
+B|1
+C|1}
+
 do_execsql_test_on_specific_db {:memory:} matview-basic-filter-population {
     CREATE TABLE products(id INTEGER, name TEXT, price INTEGER, category TEXT);
     INSERT INTO products VALUES