core/vdbe: Add NotFound bytecode

If P4==0 then register P3 holds a blob constructed by MakeRecord. If P4>0 then register P3 is the first of P4 registers that form an unpacked record.

Cursor P1 is on an index btree. If the record identified by P3 and P4 is not the prefix of any entry in P1 then a jump is made to P2. If P1 does contain an entry whose prefix matches the P3/P4 record then control falls through to the next instruction and P1 is left pointing at the matching entry.

This operation leaves the cursor in a state where it cannot be advanced in either direction. In other words, the Next and Prev opcodes do not work after this operation.

core/vdbe/execute.rs

@@ -4472,6 +4472,53 @@ pub fn op_once(
     Ok(InsnFunctionStepResult::Step)
 }
 
+pub fn op_not_found(
+    program: &Program,
+    state: &mut ProgramState,
+    insn: &Insn,
+    pager: &Rc<Pager>,
+    mv_store: Option<&Rc<MvStore>>,
+) -> Result<InsnFunctionStepResult> {
+    let Insn::NotFound {
+        cursor_id,
+        target_pc,
+        record_reg,
+        num_regs,
+    } = insn
+    else {
+        unreachable!("unexpected Insn {:?}", insn)
+    };
+
+    let found = {
+        let mut cursor = state.get_cursor(*cursor_id);
+        let cursor = cursor.as_btree_mut();
+
+        if *num_regs == 0 {
+            let record = match &state.registers[*record_reg] {
+                Register::Record(r) => r,
+                _ => {
+                    return Err(LimboError::InternalError(
+                        "NotFound: exepected a record in the register".into(),
+                    ));
+                }
+            };
+
+            return_if_io!(cursor.seek(SeekKey::IndexKey(&record), SeekOp::EQ))
+        } else {
+            let record = make_record(&state.registers, record_reg, num_regs);
+            return_if_io!(cursor.seek(SeekKey::IndexKey(&record), SeekOp::EQ))
+        }
+    };
+    
+    if found {
+        state.pc += 1;
+    } else {
+        state.pc = target_pc.to_offset_int();
+    }
+
+    Ok(InsnFunctionStepResult::Step)
+}
+
 fn exec_lower(reg: &OwnedValue) -> Option<OwnedValue> {
     match reg {
         OwnedValue::Text(t) => Some(OwnedValue::build_text(&t.as_str().to_lowercase())),

core/vdbe/explain.rs

@@ -1361,6 +1361,24 @@ pub fn insn_to_str(
                     format!("r[{}..{}]=NULL", dest, end)
                 }),
             ),
+            Insn::NotFound {
+                cursor_id,
+                target_pc,
+                record_reg,
+                ..
+            } => (
+                "NotFound",
+                *cursor_id as i32,
+                target_pc.to_debug_int(),
+                *record_reg as i32,
+                OwnedValue::build_text(""),
+                0,
+                format!(
+                    "if (r[{}] != NULL) goto {}",
+                    record_reg,
+                    target_pc.to_debug_int()
+                ),
+            ),
         };
     format!(
         "{:<4}  {:<17}  {:<4}  {:<4}  {:<4}  {:<13}  {:<2}  {}",

core/vdbe/insn.rs

@@ -802,6 +802,13 @@ pub enum Insn {
     Once {
         target_pc_when_reentered: BranchOffset,
     },
+    /// Search for record in the index cusor, if exists is a no-op otherwise go to target_pc
+    NotFound {
+        cursor_id: CursorID,
+        target_pc: BranchOffset,
+        record_reg: usize,
+        num_regs: usize,
+    },
 }
 
 impl Insn {
@@ -916,6 +923,7 @@ impl Insn {
             Insn::ReadCookie { .. } => execute::op_read_cookie,
             Insn::OpenEphemeral { .. } | Insn::OpenAutoindex { .. } => execute::op_open_ephemeral,
             Insn::Once { .. } => execute::op_once,
+            Insn::NotFound { .. } => execute::op_not_found,
         }
     }
 }