Fix cases where Insn::Insert needs to seek to ensure correct insertion

core/translate/emitter.rs

@@ -1153,7 +1153,13 @@ fn emit_update_insns(
             cursor: cursor_id,
             key_reg: rowid_set_clause_reg.unwrap_or(beg),
             record_reg,
-            flag: InsertFlags::new().update(true),
+            flag: if has_user_provided_rowid {
+                // The previous Insn::NotExists and Insn::Delete seek to the old rowid,
+                // so to insert a new user-provided rowid, we need to seek to the correct place.
+                InsertFlags::new().require_seek()
+            } else {
+                InsertFlags::new()
+            },
             table_name: table_ref.identifier.clone(),
         });
 

core/translate/insert.rs

@@ -231,7 +231,8 @@ pub fn translate_insert(
                         cursor: temp_cursor_id,
                         key_reg: rowid_reg,
                         record_reg,
-                        flag: InsertFlags::new(),
+                        // since we are not doing an Insn::NewRowid or an Insn::NotExists here, we need to seek to ensure the insertion happens in the correct place.
+                        flag: InsertFlags::new().require_seek(),
                         table_name: "".to_string(),
                     });
 

core/translate/result_row.rs

@@ -128,7 +128,8 @@ pub fn emit_result_row_and_limit(
                 cursor: *table_cursor_id,
                 key_reg: result_columns_start_reg + (plan.result_columns.len() - 1), // Rowid reg is the last register
                 record_reg,
-                flag: InsertFlags(0),
+                // since we are not doing an Insn::NewRowid or an Insn::NotExists here, we need to seek to ensure the insertion happens in the correct place.
+                flag: InsertFlags::new().require_seek(),
                 table_name: table.name.clone(),
             });
         }

core/vdbe/execute.rs

@@ -13,6 +13,7 @@ use crate::types::{
     compare_immutable, compare_records_generic, ImmutableRecord, SeekResult, Text, TextSubtype,
 };
 use crate::util::normalize_ident;
+use crate::vdbe::insn::InsertFlags;
 use crate::vdbe::registers_to_ref_values;
 use crate::{
     error::{
@@ -4884,11 +4885,15 @@ pub fn op_insert(
             Register::Aggregate(..) => unreachable!("Cannot insert an aggregate value."),
         };
 
-        // In a table insert, we can always assume we have moved to the correct place, because
-        // 1. if the rowid is autogenerated, op_new_rowid() always gets called and we seek to the end of the table.
-        // 2. if the rowid is not autogenerated, op_not_exists() always gets called to ensure the rowid being inserted does not exist.
-        // Both of these code paths perform a seek, which lands us at the correct insertion place.
-        return_if_io!(cursor.insert(&BTreeKey::new_table_rowid(key, Some(record.as_ref())), true));
+        // In a table insert, if the caller does not pass InsertFlags::REQUIRE_SEEK, they must ensure that a seek has already happened to the correct location.
+        // This typically happens by invoking either Insn::NewRowid or Insn::NotExists, because:
+        // 1. op_new_rowid() seeks to the end of the table, which is the correct insertion position.
+        // 2. op_not_exists() seeks to the position in the table where the target rowid would be inserted.
+        let moved_before = !flag.has(InsertFlags::REQUIRE_SEEK);
+        return_if_io!(cursor.insert(
+            &BTreeKey::new_table_rowid(key, Some(record.as_ref())),
+            moved_before
+        ));
     }
 
     // Only update last_insert_rowid for regular table inserts, not schema modifications

core/vdbe/insn.rs

@@ -109,6 +109,7 @@ pub struct InsertFlags(pub u8);
 
 impl InsertFlags {
     pub const UPDATE: u8 = 0x01; // Flag indicating this is part of an UPDATE statement
+    pub const REQUIRE_SEEK: u8 = 0x02; // Flag indicating that a seek is required to insert the row
 
     pub fn new() -> Self {
         InsertFlags(0)
@@ -118,12 +119,8 @@ impl InsertFlags {
         (self.0 & flag) != 0
     }
 
-    pub fn update(mut self, is_update: bool) -> Self {
-        if is_update {
-            self.0 |= InsertFlags::UPDATE;
-        } else {
-            self.0 &= !InsertFlags::UPDATE;
-        }
+    pub fn require_seek(mut self) -> Self {
+        self.0 |= InsertFlags::REQUIRE_SEEK;
         self
     }
 }