return Error from step if IO failed

core/io/mod.rs

@@ -214,6 +214,10 @@ impl Completion {
         self.inner.result.get().is_some_and(|val| val.is_some())
     }
 
+    pub fn get_error(&self) -> Option<CompletionError> {
+        self.inner.result.get().and_then(|res| *res)
+    }
+
     /// Checks if the Completion completed or errored
     pub fn finished(&self) -> bool {
         self.inner.result.get().is_some()

core/types.rs

@@ -14,7 +14,7 @@ use crate::translate::plan::IterationDirection;
 use crate::vdbe::sorter::Sorter;
 use crate::vdbe::Register;
 use crate::vtab::VirtualTableCursor;
-use crate::{turso_assert, Completion, Result, IO};
+use crate::{turso_assert, Completion, CompletionError, Result, IO};
 use std::fmt::{Debug, Display};
 
 const MAX_REAL_SIZE: u8 = 15;
@@ -2502,6 +2502,13 @@ impl IOCompletions {
             IOCompletions::Many(completions) => completions.iter().for_each(|c| c.abort()),
         }
     }
+
+    pub fn get_error(&self) -> Option<CompletionError> {
+        match self {
+            IOCompletions::Single(c) => c.get_error(),
+            IOCompletions::Many(completions) => completions.iter().find_map(|c| c.get_error()),
+        }
+    }
 }
 
 #[derive(Debug)]

core/vdbe/mod.rs

@@ -460,6 +460,11 @@ impl Program {
                 if !io.finished() {
                     return Ok(StepResult::IO);
                 }
+                if let Some(err) = io.get_error() {
+                    let err = err.into();
+                    handle_program_error(&pager, &self.connection, &err)?;
+                    return Err(err);
+                }
                 state.io_completions = None;
             }
             // invalidate row