Merge branch 'main' into json-extract

CONTRIBUTING.md

@@ -17,6 +17,10 @@ If you are new to Rust, the following books are recommended reading:
 * Jim Blandy et al. [Programming Rust, 2nd Edition](https://www.oreilly.com/library/view/programming-rust-2nd/9781492052586/). 2021
 * Steve Klabnik and Carol Nichols. [The Rust Programming Language](https://doc.rust-lang.org/book/#the-rust-programming-language). 2022
 
+Examples of contributing
+
+* [How to contribute a SQL function implementation](docs/internals/functions.md)
+
 ## Finding things to work on
 
 The issue tracker has issues tagged with [good first issue](https://github.com/penberg/limbo/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22),

bindings/python/build.rs

@@ -1,4 +1,3 @@
 fn main() {
     pyo3_build_config::use_pyo3_cfgs();
-    println!("cargo::rustc-check-cfg=cfg(allocator, values(\"default\", \"mimalloc\"))");
 }

bindings/python/src/lib.rs

@@ -1,11 +1,11 @@
 use anyhow::Result;
 use errors::*;
-use limbo_core::IO;
 use pyo3::prelude::*;
 use pyo3::types::PyList;
 use pyo3::types::PyTuple;
+use std::cell::RefCell;
 use std::rc::Rc;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 mod errors;
 
@@ -78,7 +78,7 @@ pub struct Cursor {
     #[pyo3(get)]
     rowcount: i64,
 
-    smt: Option<Arc<Mutex<limbo_core::Statement>>>,
+    smt: Option<Rc<RefCell<limbo_core::Statement>>>,
 }
 
 // SAFETY: The limbo_core crate guarantees that `Cursor` is thread-safe.
@@ -90,26 +90,33 @@ impl Cursor {
     #[pyo3(signature = (sql, parameters=None))]
     pub fn execute(&mut self, sql: &str, parameters: Option<Py<PyTuple>>) -> Result<Self> {
         let stmt_is_dml = stmt_is_dml(sql);
+        let stmt_is_ddl = stmt_is_ddl(sql);
 
-        let conn_lock =
-            self.conn.conn.lock().map_err(|_| {
-                PyErr::new::<OperationalError, _>("Failed to acquire connection lock")
-            })?;
-
-        let statement = conn_lock.prepare(sql).map_err(|e| {
+        let statement = self.conn.conn.prepare(sql).map_err(|e| {
             PyErr::new::<ProgrammingError, _>(format!("Failed to prepare statement: {:?}", e))
         })?;
 
-        self.smt = Some(Arc::new(Mutex::new(statement)));
+        let stmt = Rc::new(RefCell::new(statement));
 
-        // TODO: use stmt_is_dml to set rowcount
-        if stmt_is_dml {
-            return Err(PyErr::new::<NotSupportedError, _>(
-                "DML statements (INSERT/UPDATE/DELETE) are not fully supported in this version",
-            )
-            .into());
+        // For DDL and DML statements,
+        // we need to execute the statement immediately
+        if stmt_is_ddl || stmt_is_dml {
+            loop {
+                match stmt.borrow_mut().step().map_err(|e| {
+                    PyErr::new::<OperationalError, _>(format!("Step error: {:?}", e))
+                })? {
+                    limbo_core::StepResult::IO => {
+                        self.conn.io.run_once().map_err(|e| {
+                            PyErr::new::<OperationalError, _>(format!("IO error: {:?}", e))
+                        })?;
+                    }
+                    _ => break,
+                }
+            }
         }
 
+        self.smt = Some(stmt);
+
         Ok(Cursor {
             smt: self.smt.clone(),
             conn: self.conn.clone(),
@@ -121,11 +128,8 @@ impl Cursor {
 
     pub fn fetchone(&mut self, py: Python) -> Result<Option<PyObject>> {
         if let Some(smt) = &self.smt {
-            let mut smt_lock = smt.lock().map_err(|_| {
-                PyErr::new::<OperationalError, _>("Failed to acquire statement lock")
-            })?;
             loop {
-                match smt_lock.step().map_err(|e| {
+                match smt.borrow_mut().step().map_err(|e| {
                     PyErr::new::<OperationalError, _>(format!("Step error: {:?}", e))
                 })? {
                     limbo_core::StepResult::Row(row) => {
@@ -157,14 +161,9 @@ impl Cursor {
 
     pub fn fetchall(&mut self, py: Python) -> Result<Vec<PyObject>> {
         let mut results = Vec::new();
-
         if let Some(smt) = &self.smt {
-            let mut smt_lock = smt.lock().map_err(|_| {
-                PyErr::new::<OperationalError, _>("Failed to acquire statement lock")
-            })?;
-
             loop {
-                match smt_lock.step().map_err(|e| {
+                match smt.borrow_mut().step().map_err(|e| {
                     PyErr::new::<OperationalError, _>(format!("Step error: {:?}", e))
                 })? {
                     limbo_core::StepResult::Row(row) => {
@@ -221,11 +220,17 @@ fn stmt_is_dml(sql: &str) -> bool {
     sql.starts_with("INSERT") || sql.starts_with("UPDATE") || sql.starts_with("DELETE")
 }
 
+fn stmt_is_ddl(sql: &str) -> bool {
+    let sql = sql.trim();
+    let sql = sql.to_uppercase();
+    sql.starts_with("CREATE") || sql.starts_with("ALTER") || sql.starts_with("DROP")
+}
+
 #[pyclass]
 #[derive(Clone)]
 pub struct Connection {
-    conn: Arc<Mutex<Rc<limbo_core::Connection>>>,
-    io: Arc<limbo_core::PlatformIO>,
+    conn: Rc<limbo_core::Connection>,
+    io: Arc<dyn limbo_core::IO>,
 }
 
 // SAFETY: The limbo_core crate guarantees that `Connection` is thread-safe.
@@ -263,16 +268,24 @@ impl Connection {
 #[allow(clippy::arc_with_non_send_sync)]
 #[pyfunction]
 pub fn connect(path: &str) -> Result<Connection> {
-    let io = Arc::new(limbo_core::PlatformIO::new().map_err(|e| {
-        PyErr::new::<InterfaceError, _>(format!("IO initialization failed: {:?}", e))
-    })?);
-    let db = limbo_core::Database::open_file(io.clone(), path)
-        .map_err(|e| PyErr::new::<DatabaseError, _>(format!("Failed to open database: {:?}", e)))?;
-    let conn: Rc<limbo_core::Connection> = db.connect();
-    Ok(Connection {
-        conn: Arc::new(Mutex::new(conn)),
-        io,
-    })
+    match path {
+        ":memory:" => {
+            let io: Arc<dyn limbo_core::IO> = Arc::new(limbo_core::MemoryIO::new()?);
+            let db = limbo_core::Database::open_file(io.clone(), path).map_err(|e| {
+                PyErr::new::<DatabaseError, _>(format!("Failed to open database: {:?}", e))
+            })?;
+            let conn: Rc<limbo_core::Connection> = db.connect();
+            Ok(Connection { conn, io })
+        }
+        path => {
+            let io: Arc<dyn limbo_core::IO> = Arc::new(limbo_core::PlatformIO::new()?);
+            let db = limbo_core::Database::open_file(io.clone(), path).map_err(|e| {
+                PyErr::new::<DatabaseError, _>(format!("Failed to open database: {:?}", e))
+            })?;
+            let conn: Rc<limbo_core::Connection> = db.connect();
+            Ok(Connection { conn, io })
+        }
+    }
 }
 
 fn row_to_py(py: Python, row: &limbo_core::Row) -> PyObject {

bindings/python/tests/test_database.py

@@ -27,6 +27,20 @@ def test_fetchall_select_user_ids(provider):
     assert user_ids == [(1,), (2,)]
 
 
+@pytest.mark.parametrize("provider", ["sqlite3", "limbo"])
+def test_in_memory_fetchone_select_all_users(provider):
+    conn = connect(provider, ":memory:")
+    cursor = conn.cursor()
+    cursor.execute("CREATE TABLE users (id INT PRIMARY KEY, username TEXT)")
+    cursor.execute("INSERT INTO users VALUES (1, 'alice')")
+
+    cursor.execute("SELECT * FROM users")
+
+    alice = cursor.fetchone()
+    assert alice
+    assert alice == (1, "alice")
+
+
 @pytest.mark.parametrize("provider", ["sqlite3", "limbo"])
 def test_fetchone_select_all_users(provider):
     conn = connect(provider, "tests/database.db")

core/function.rs

@@ -17,9 +17,9 @@ impl Display for JsonFunc {
             f,
             "{}",
             match self {
-                JsonFunc::Json => "json".to_string(),
-                JsonFunc::JsonArray => "json_array".to_string(),
-                JsonFunc::JsonExtract => "json_extract".to_string(),
+                Self::Json => "json".to_string(),
+                Self::JsonArray => "json_array".to_string(),
+                Self::JsonExtract => "json_extract".to_string(),
                 Self::JsonArrayLength => "json_array_length".to_string(),
             }
         )

core/io/linux.rs

@@ -11,6 +11,7 @@ use std::rc::Rc;
 use thiserror::Error;
 
 const MAX_IOVECS: usize = 128;
+const SQPOLL_IDLE: u32 = 1000;
 
 #[derive(Debug, Error)]
 enum LinuxIOError {
@@ -49,7 +50,13 @@ struct InnerLinuxIO {
 
 impl LinuxIO {
     pub fn new() -> Result<Self> {
-        let ring = io_uring::IoUring::new(MAX_IOVECS as u32)?;
+        let ring = match io_uring::IoUring::builder()
+            .setup_sqpoll(SQPOLL_IDLE)
+            .build(MAX_IOVECS as u32)
+        {
+            Ok(ring) => ring,
+            Err(_) => io_uring::IoUring::new(MAX_IOVECS as u32)?,
+        };
         let inner = InnerLinuxIO {
             ring: WrappedIOUring {
                 ring,

core/json/de.rs

@@ -41,7 +41,7 @@ impl<'de> Deserializer<'de> {
     }
 }
 
-impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
+impl<'de> de::Deserializer<'de> for &mut Deserializer<'de> {
     type Error = Error;
 
     fn deserialize_any<V>(self, visitor: V) -> Result<V::Value>

core/json/mod.rs

@@ -66,7 +66,7 @@ fn get_json_value(json_value: &OwnedValue) -> crate::Result<Val> {
             }
         },
         OwnedValue::Blob(b) => {
-            if let Ok(json) = jsonb::from_slice(b) {
+            if let Ok(_json) = jsonb::from_slice(b) {
                 todo!("jsonb to json conversion");
             } else {
                 crate::bail_parse_error!("malformed JSON");
@@ -137,7 +137,7 @@ pub fn json_array_length(
     };
 
     match arr_val {
-        Val::Array(val) => (Ok(OwnedValue::Integer(val.len() as i64))),
+        Val::Array(val) => Ok(OwnedValue::Integer(val.len() as i64)),
         Val::Null => Ok(OwnedValue::Null),
         _ => Ok(OwnedValue::Integer(0)),
     }

core/json/ser.rs

@@ -30,7 +30,7 @@ impl Serializer {
     }
 }
 
-impl<'a> ser::Serializer for &'a mut Serializer {
+impl ser::Serializer for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -237,7 +237,7 @@ impl<'a> ser::Serializer for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeSeq for &'a mut Serializer {
+impl ser::SerializeSeq for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -257,7 +257,7 @@ impl<'a> ser::SerializeSeq for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeTuple for &'a mut Serializer {
+impl ser::SerializeTuple for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -273,7 +273,7 @@ impl<'a> ser::SerializeTuple for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeTupleStruct for &'a mut Serializer {
+impl ser::SerializeTupleStruct for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -289,7 +289,7 @@ impl<'a> ser::SerializeTupleStruct for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeTupleVariant for &'a mut Serializer {
+impl ser::SerializeTupleVariant for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -306,7 +306,7 @@ impl<'a> ser::SerializeTupleVariant for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeMap for &'a mut Serializer {
+impl ser::SerializeMap for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -334,7 +334,7 @@ impl<'a> ser::SerializeMap for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeStruct for &'a mut Serializer {
+impl ser::SerializeStruct for &mut Serializer {
     type Ok = ();
     type Error = Error;
 
@@ -351,7 +351,7 @@ impl<'a> ser::SerializeStruct for &'a mut Serializer {
     }
 }
 
-impl<'a> ser::SerializeStructVariant for &'a mut Serializer {
+impl ser::SerializeStructVariant for &mut Serializer {
     type Ok = ();
     type Error = Error;
 

core/storage/btree.rs

@@ -1378,10 +1378,10 @@ impl BTreeCursor {
                     PageType::IndexLeaf => todo!(),
                 };
                 cbrk -= size;
-                if cbrk < first_cell as u64 || pc + size > usable_space {
+                if cbrk < first_cell || pc + size > usable_space {
                     todo!("corrupt");
                 }
-                assert!(cbrk + size <= usable_space && cbrk >= first_cell as u64);
+                assert!(cbrk + size <= usable_space && cbrk >= first_cell);
                 // set new pointer
                 write_buf[cell_idx..cell_idx + 2].copy_from_slice(&(cbrk as u16).to_be_bytes());
                 // copy payload
@@ -1394,7 +1394,7 @@ impl BTreeCursor {
         // if( data[hdr+7]+cbrk-iCellFirst!=pPage->nFree ){
         //   return SQLITE_CORRUPT_PAGE(pPage);
         // }
-        assert!(cbrk >= first_cell as u64);
+        assert!(cbrk >= first_cell);
         let write_buf = page.as_ptr();
 
         // set new first byte of cell content
@@ -1437,7 +1437,7 @@ impl BTreeCursor {
         // #3. freeblocks (linked list of blocks of at least 4 bytes within the cell content area that are not in use due to e.g. deletions)
 
         let mut free_space_bytes =
-            page.unallocated_region_size() as usize + page.num_frag_free_bytes() as usize;
+            page.unallocated_region_size() + page.num_frag_free_bytes() as usize;
 
         // #3 is computed by iterating over the freeblocks linked list
         let mut cur_freeblock_ptr = page.first_freeblock() as usize;