aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-02-23 17:05:36 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge 'Initial pass on Virtual FileSystem extension module' from Preston Thorpe

Browse Source 
This PR implements a `VFS` module for our extension library, allowing
extensions to be written that can introduce different I/O back-ends.
EDIT: there is an included plain external/sync vfs example for testing,
as mentioned in #996 they can be combined after they are both merged and
we can keep 1 extension crate just for testing that features can be
added to, without making a new extension just to test stuff.
This PR also adds the `.vfslist` dot command, and replaces the `--io`
CLI argument with `--vfs` to match sqlite.
In order to support building vfs modules at compile time, and to then
support opening a brand new db file using a staticly built-in extension
module, a new method was created `open_with_vfs` that will load any vfs
modules before a `Database` is created, and uses that `IO` to create the
initial file, and returns it: `Result<(Arc<dyn IO>, Arc<Database>)>`. in
keeping with the API of core.
When #1039 is merged, the vfs module can be specified in a query
parameter.

Closes #960
This commit is contained in:
Pekka Enberg
2025-03-13 21:16:52 +02:00
parent 9f641f17c6 c638b64a59
commit b1419ab434
21 changed files with 1240 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
Cargo.lock generated
View File

@@ -336,16 +336,16 @@ checksum = "18758054972164c3264f7c8386f5fc6da6114cb46b619fd365d4e3b2dc3ae487"
[[package]]
name = "chrono"
version = "0.4.39"
version = "0.4.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825"
checksum = "1a7964611d71df112cb1730f2ee67324fcf4d0fc6606acbbe9bfe06df124637c"
dependencies = [
"android-tzdata",
"iana-time-zone",
"js-sys",
"num-traits",
"wasm-bindgen",
"windows-targets 0.52.6",
"windows-link",
]
[[package]]
@@ -1658,6 +1658,7 @@ dependencies = [
"limbo_completion",
"limbo_crypto",
"limbo_ext",
"limbo_ext_tests",
"limbo_ipaddr",
"limbo_macros",
"limbo_percentile",
@@ -1709,9 +1710,22 @@ dependencies = [
name = "limbo_ext"
version = "0.0.16"
dependencies = [
"chrono",
"getrandom 0.3.1",
"limbo_macros",
]
[[package]]
name = "limbo_ext_tests"
version = "0.0.16"
dependencies = [
"env_logger 0.11.6",
"lazy_static",
"limbo_ext",
"log",
"mimalloc",
]
[[package]]
name = "limbo_ipaddr"
version = "0.0.16"
@@ -1721,15 +1735,6 @@ dependencies = [
"mimalloc",
]
[[package]]
name = "limbo_kv"
version = "0.0.16"
dependencies = [
"lazy_static",
"limbo_ext",
"mimalloc",
]
[[package]]
name = "limbo_macros"
version = "0.0.16"
@@ -1892,9 +1897,9 @@ dependencies = [
[[package]]
name = "log"
version = "0.4.25"
version = "0.4.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04cbf5b083de1c7e0222a7a51dbfdba1cbe1c6ab0b15e29fff3f6c077fd9cd9f"
checksum = "30bde2b3dc3671ae49d8e2e9f044c7c005836e7a023ee57cffa25ab82764bb9e"
[[package]]
name = "lru"
@@ -3614,6 +3619,12 @@ dependencies = [
"windows-targets 0.52.6",
]
[[package]]
name = "windows-link"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dccfd733ce2b1753b03b6d3c65edf020262ea35e20ccdf3e288043e6dd620e3"
[[package]]
name = "windows-sys"
version = "0.45.0"

3
Cargo.toml
View File

@@ -13,7 +13,7 @@ members = [
"extensions/completion",
"extensions/core",
"extensions/crypto",
"extensions/kvstore",
"extensions/tests",
"extensions/percentile",
"extensions/regexp",
"extensions/series",
@@ -47,6 +47,7 @@ limbo_uuid = { path = "extensions/uuid", version = "0.0.16" }
limbo_sqlite3_parser = { path = "vendored/sqlite3-parser", version = "0.0.16" }
limbo_ipaddr = { path = "extensions/ipaddr", version = "0.0.16" }
limbo_completion = { path = "extensions/completion", version = "0.0.16" }
limbo_ext_tests = { path = "extensions/tests", version = "0.0.16" }
# Config for 'cargo dist'
[workspace.metadata.dist]

75
cli/app.rs
View File

@@ -1,7 +1,7 @@
use crate::{
helper::LimboHelper,
import::{ImportFile, IMPORT_HELP},
input::{get_io, get_writer, DbLocation, Io, OutputMode, Settings, HELP_MSG},
input::{get_io, get_writer, DbLocation, OutputMode, Settings, HELP_MSG},
opcodes_dictionary::OPCODE_DESCRIPTIONS,
};
use comfy_table::{Attribute, Cell, CellAlignment, ContentArrangement, Row, Table};
@@ -43,14 +43,11 @@ pub struct Opts {
#[clap(short, long, help = "Print commands before execution")]
pub echo: bool,
#[clap(
default_value_t,
value_enum,
short,
short = 'v',
long,
help = "Select I/O backend. The only other choice to 'syscall' is\n\
\t'io-uring' when built for Linux with feature 'io_uring'\n"
help = "Select VFS. options are io_uring (if feature enabled), memory, and syscall"
)]
pub io: Io,
pub vfs: Option<String>,
#[clap(long, help = "Enable experimental MVCC feature")]
pub experimental_mvcc: bool,
}
@@ -89,6 +86,8 @@ pub enum Command {
LoadExtension,
/// Dump the current database as a list of SQL statements
Dump,
/// List vfs modules available
ListVfs,
}
impl Command {
@@ -102,6 +101,7 @@ impl Command {
| Self::ShowInfo
| Self::Tables
| Self::SetOutput
| Self::ListVfs
| Self::Dump => 0,
Self::Open
| Self::OutputMode
@@ -131,6 +131,7 @@ impl Command {
Self::LoadExtension => ".load",
Self::Dump => ".dump",
Self::Import => &IMPORT_HELP,
Self::ListVfs => ".vfslist",
}
}
}
@@ -155,6 +156,7 @@ impl FromStr for Command {
".import" => Ok(Self::Import),
".load" => Ok(Self::LoadExtension),
".dump" => Ok(Self::Dump),
".vfslist" => Ok(Self::ListVfs),
_ => Err("Unknown command".to_string()),
}
}
@@ -205,15 +207,27 @@ impl<'a> Limbo<'a> {
.database
.as_ref()
.map_or(":memory:".to_string(), |p| p.to_string_lossy().to_string());
let io = {
match db_file.as_str() {
":memory:" => get_io(DbLocation::Memory, opts.io)?,
_path => get_io(DbLocation::Path, opts.io)?,
}
let (io, db) = if let Some(ref vfs) = opts.vfs {
Database::open_new(&db_file, vfs)?
} else {
let io = {
match db_file.as_str() {
":memory:" => get_io(
DbLocation::Memory,
opts.vfs.as_ref().map_or("", |s| s.as_str()),
)?,
_path => get_io(
DbLocation::Path,
opts.vfs.as_ref().map_or("", |s| s.as_str()),
)?,
}
};
(
io.clone(),
Database::open_file(io.clone(), &db_file, opts.experimental_mvcc)?,
)
};
let db = Database::open_file(io.clone(), &db_file, opts.experimental_mvcc)?;
let conn = db.connect().unwrap();
let conn = db.connect()?;
let h = LimboHelper::new(conn.clone(), io.clone());
rl.set_helper(Some(h));
let interrupt_count = Arc::new(AtomicUsize::new(0));
@@ -405,17 +419,21 @@ impl<'a> Limbo<'a> {
}
}
fn open_db(&mut self, path: &str) -> anyhow::Result<()> {
fn open_db(&mut self, path: &str, vfs_name: Option<&str>) -> anyhow::Result<()> {
self.conn.close()?;
let io = {
match path {
":memory:" => get_io(DbLocation::Memory, self.opts.io)?,
_path => get_io(DbLocation::Path, self.opts.io)?,
}
let (io, db) = if let Some(vfs_name) = vfs_name {
self.conn.open_new(path, vfs_name)?
} else {
let io = {
match path {
":memory:" => get_io(DbLocation::Memory, &self.opts.io.to_string())?,
_path => get_io(DbLocation::Path, &self.opts.io.to_string())?,
}
};
(io.clone(), Database::open_file(io.clone(), path, false)?)
};
self.io = Arc::clone(&io);
let db = Database::open_file(self.io.clone(), path, self.opts.experimental_mvcc)?;
self.conn = db.connect().unwrap();
self.io = io;
self.conn = db.connect()?;
self.opts.db_file = path.to_string();
Ok(())
}
@@ -569,7 +587,8 @@ impl<'a> Limbo<'a> {
std::process::exit(0)
}
Command::Open => {
if self.open_db(args[1]).is_err() {
let vfs = args.get(2).map(|s| &**s);
if self.open_db(args[1], vfs).is_err() {
let _ = self.writeln("Error: Unable to open database file.");
}
}
@@ -651,6 +670,12 @@ impl<'a> Limbo<'a> {
let _ = self.write_fmt(format_args!("/****** ERROR: {} ******/", e));
}
}
Command::ListVfs => {
let _ = self.writeln("Available VFS modules:");
self.conn.list_vfs().iter().for_each(|v| {
let _ = self.writeln(v);
});
}
}
} else {
let _ = self.write_fmt(format_args!(

37
cli/input.rs
View File

@@ -1,6 +1,7 @@
use crate::app::Opts;
use clap::ValueEnum;
use std::{
fmt::{Display, Formatter},
io::{self, Write},
sync::Arc,
};
@@ -11,11 +12,26 @@ pub enum DbLocation {
Path,
}
#[derive(Copy, Clone, ValueEnum)]
#[allow(clippy::enum_variant_names)]
#[derive(Clone, Debug)]
pub enum Io {
Syscall,
#[cfg(all(target_os = "linux", feature = "io_uring"))]
IoUring,
External(String),
Memory,
}
impl Display for Io {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Io::Memory => write!(f, "memory"),
Io::Syscall => write!(f, "syscall"),
#[cfg(all(target_os = "linux", feature = "io_uring"))]
Io::IoUring => write!(f, "io_uring"),
Io::External(str) => write!(f, "{}", str),
}
}
}
impl Default for Io {
@@ -65,7 +81,6 @@ pub struct Settings {
pub echo: bool,
pub is_stdout: bool,
pub io: Io,
pub experimental_mvcc: bool,
}
impl From<&Opts> for Settings {
@@ -80,8 +95,14 @@ impl From<&Opts> for Settings {
.database
.as_ref()
.map_or(":memory:".to_string(), |p| p.to_string_lossy().to_string()),
io: opts.io,
experimental_mvcc: opts.experimental_mvcc,
io: match opts.vfs.as_ref().unwrap_or(&String::new()).as_str() {
"memory" => Io::Memory,
"syscall" => Io::Syscall,
#[cfg(all(target_os = "linux", feature = "io_uring"))]
"io_uring" => Io::IoUring,
"" => Io::default(),
vfs => Io::External(vfs.to_string()),
},
}
}
}
@@ -120,12 +141,13 @@ pub fn get_writer(output: &str) -> Box<dyn Write> {
}
}
pub fn get_io(db_location: DbLocation, io_choice: Io) -> anyhow::Result<Arc<dyn limbo_core::IO>> {
pub fn get_io(db_location: DbLocation, io_choice: &str) -> anyhow::Result<Arc<dyn limbo_core::IO>> {
Ok(match db_location {
DbLocation::Memory => Arc::new(limbo_core::MemoryIO::new()),
DbLocation::Path => {
match io_choice {
Io::Syscall => {
"memory" => Arc::new(limbo_core::MemoryIO::new()),
"syscall" => {
// We are building for Linux/macOS and syscall backend has been selected
#[cfg(target_family = "unix")]
{
@@ -139,7 +161,8 @@ pub fn get_io(db_location: DbLocation, io_choice: Io) -> anyhow::Result<Arc<dyn
}
// We are building for Linux and io_uring backend has been selected
#[cfg(all(target_os = "linux", feature = "io_uring"))]
Io::IoUring => Arc::new(limbo_core::UringIO::new()?),
"io_uring" => Arc::new(limbo_core::UringIO::new()?),
_ => Arc::new(limbo_core::PlatformIO::new()?),
}
}
})

2
core/Cargo.toml
View File

@@ -26,6 +26,7 @@ crypto = ["limbo_crypto/static"]
series = ["limbo_series/static"]
ipaddr = ["limbo_ipaddr/static"]
completion = ["limbo_completion/static"]
testvfs = ["limbo_ext_tests/static"]
[target.'cfg(target_os = "linux")'.dependencies]
io-uring = { version = "0.6.1", optional = true }
@@ -68,6 +69,7 @@ limbo_crypto = { workspace = true, optional = true, features = ["static"] }
limbo_series = { workspace = true, optional = true, features = ["static"] }
limbo_ipaddr = { workspace = true, optional = true, features = ["static"] }
limbo_completion = { workspace = true, optional = true, features = ["static"] }
limbo_ext_tests = { workspace = true, optional = true, features = ["static"] }
miette = "7.4.0"
strum = "0.26"
parking_lot = "0.12.3"

41
core/ext/dynamic.rs Normal file
View File

@@ -0,0 +1,41 @@
use crate::{Connection, LimboError};
use libloading::{Library, Symbol};
use limbo_ext::{ExtensionApi, ExtensionApiRef, ExtensionEntryPoint};
use std::sync::{Arc, Mutex, OnceLock};
type ExtensionStore = Vec<(Arc<Library>, ExtensionApiRef)>;
static EXTENSIONS: OnceLock<Arc<Mutex<ExtensionStore>>> = OnceLock::new();
pub fn get_extension_libraries() -> Arc<Mutex<ExtensionStore>> {
EXTENSIONS
.get_or_init(|| Arc::new(Mutex::new(Vec::new())))
.clone()
}
impl Connection {
pub fn load_extension<P: AsRef<std::ffi::OsStr>>(&self, path: P) -> crate::Result<()> {
use limbo_ext::ExtensionApiRef;
let api = Box::new(self.build_limbo_ext());
let lib =
unsafe { Library::new(path).map_err(|e| LimboError::ExtensionError(e.to_string()))? };
let entry: Symbol<ExtensionEntryPoint> = unsafe {
lib.get(b"register_extension")
.map_err(|e| LimboError::ExtensionError(e.to_string()))?
};
let api_ptr: *const ExtensionApi = Box::into_raw(api);
let api_ref = ExtensionApiRef { api: api_ptr };
let result_code = unsafe { entry(api_ptr) };
if result_code.is_ok() {
let extensions = get_extension_libraries();
extensions.lock().unwrap().push((Arc::new(lib), api_ref));
Ok(())
} else {
if !api_ptr.is_null() {
let _ = unsafe { Box::from_raw(api_ptr.cast_mut()) };
}
Err(LimboError::ExtensionError(
"Extension registration failed".to_string(),
))
}
}
}

179
core/ext/mod.rs
View File

@@ -1,13 +1,22 @@
use crate::{function::ExternalFunc, Connection};
#[cfg(not(target_family = "wasm"))]
mod dynamic;
#[cfg(all(target_os = "linux", feature = "io_uring"))]
use crate::UringIO;
use crate::IO;
use crate::{function::ExternalFunc, Connection, Database, LimboError};
use limbo_ext::{
ExtensionApi, InitAggFunction, ResultCode, ScalarFunction, VTabKind, VTabModuleImpl,
ExtensionApi, InitAggFunction, ResultCode, ScalarFunction, VTabKind, VTabModuleImpl, VfsImpl,
};
pub use limbo_ext::{FinalizeFunction, StepFunction, Value as ExtValue, ValueType as ExtValueType};
use std::{
ffi::{c_char, c_void, CStr, CString},
rc::Rc,
sync::{Arc, Mutex, OnceLock},
};
type ExternAggFunc = (InitAggFunction, StepFunction, FinalizeFunction);
type Vfs = (String, Arc<VfsMod>);
static VFS_MODULES: OnceLock<Mutex<Vec<Vfs>>> = OnceLock::new();
#[derive(Clone)]
pub struct VTabImpl {
@@ -15,6 +24,14 @@ pub struct VTabImpl {
pub implementation: Rc<VTabModuleImpl>,
}
#[derive(Clone, Debug)]
pub struct VfsMod {
pub ctx: *const VfsImpl,
}
unsafe impl Send for VfsMod {}
unsafe impl Sync for VfsMod {}
unsafe extern "C" fn register_scalar_function(
ctx: *mut c_void,
name: *const c_char,
@@ -74,6 +91,108 @@ unsafe extern "C" fn register_module(
conn.register_module_impl(&name_str, module, kind)
}
#[allow(clippy::arc_with_non_send_sync)]
unsafe extern "C" fn register_vfs(name: *const c_char, vfs: *const VfsImpl) -> ResultCode {
if name.is_null() || vfs.is_null() {
return ResultCode::Error;
}
let c_str = unsafe { CString::from_raw(name as *mut i8) };
let name_str = match c_str.to_str() {
Ok(s) => s.to_string(),
Err(_) => return ResultCode::Error,
};
add_vfs_module(name_str, Arc::new(VfsMod { ctx: vfs }));
ResultCode::OK
}
/// Get pointers to all the vfs extensions that need to be built in at compile time.
/// any other types that are defined in the same extension will not be registered
/// until the database file is opened and `register_builtins` is called.
#[cfg(feature = "fs")]
#[allow(clippy::arc_with_non_send_sync)]
pub fn add_builtin_vfs_extensions(
api: Option<ExtensionApi>,
) -> crate::Result<Vec<(String, Arc<VfsMod>)>> {
let mut vfslist: Vec<*const VfsImpl> = Vec::new();
let mut api = match api {
None => ExtensionApi {
ctx: std::ptr::null_mut(),
register_scalar_function,
register_aggregate_function,
register_vfs,
register_module,
builtin_vfs: vfslist.as_mut_ptr(),
builtin_vfs_count: 0,
},
Some(mut api) => {
api.builtin_vfs = vfslist.as_mut_ptr();
api
}
};
register_static_vfs_modules(&mut api);
let mut vfslist = Vec::with_capacity(api.builtin_vfs_count as usize);
let slice =
unsafe { std::slice::from_raw_parts_mut(api.builtin_vfs, api.builtin_vfs_count as usize) };
for vfs in slice {
if vfs.is_null() {
continue;
}
let vfsimpl = unsafe { &**vfs };
let name = unsafe {
CString::from_raw(vfsimpl.name as *mut i8)
.to_str()
.map_err(|_| {
LimboError::ExtensionError("unable to register vfs extension".to_string())
})?
.to_string()
};
vfslist.push((
name,
Arc::new(VfsMod {
ctx: vfsimpl as *const _,
}),
));
}
Ok(vfslist)
}
fn register_static_vfs_modules(_api: &mut ExtensionApi) {
#[cfg(feature = "testvfs")]
unsafe {
limbo_ext_tests::register_extension_static(_api);
}
}
impl Database {
#[cfg(feature = "fs")]
#[allow(clippy::arc_with_non_send_sync, dead_code)]
pub fn open_with_vfs(
&self,
path: &str,
vfs: &str,
) -> crate::Result<(Arc<dyn IO>, Arc<Database>)> {
use crate::{MemoryIO, PlatformIO};
let io: Arc<dyn IO> = match vfs {
"memory" => Arc::new(MemoryIO::new()),
"syscall" => Arc::new(PlatformIO::new()?),
#[cfg(all(target_os = "linux", feature = "io_uring"))]
"io_uring" => Arc::new(UringIO::new()?),
other => match get_vfs_modules().iter().find(|v| v.0 == vfs) {
Some((_, vfs)) => vfs.clone(),
None => {
return Err(LimboError::InvalidArgument(format!(
"no such VFS: {}",
other
)));
}
},
};
let db = Self::open_file(io.clone(), path, false)?;
Ok((io, db))
}
}
impl Connection {
fn register_scalar_function_impl(&self, name: &str, func: ScalarFunction) -> ResultCode {
self.syms.borrow_mut().functions.insert(
@@ -120,44 +239,82 @@ impl Connection {
register_scalar_function,
register_aggregate_function,
register_module,
register_vfs,
builtin_vfs: std::ptr::null_mut(),
builtin_vfs_count: 0,
}
}
pub fn register_builtins(&self) -> Result<(), String> {
#[allow(unused_variables)]
let ext_api = self.build_limbo_ext();
let mut ext_api = self.build_limbo_ext();
#[cfg(feature = "uuid")]
if unsafe { !limbo_uuid::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_uuid::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register uuid extension".to_string());
}
#[cfg(feature = "percentile")]
if unsafe { !limbo_percentile::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_percentile::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register percentile extension".to_string());
}
#[cfg(feature = "regexp")]
if unsafe { !limbo_regexp::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_regexp::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register regexp extension".to_string());
}
#[cfg(feature = "time")]
if unsafe { !limbo_time::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_time::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register time extension".to_string());
}
#[cfg(feature = "crypto")]
if unsafe { !limbo_crypto::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_crypto::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register crypto extension".to_string());
}
#[cfg(feature = "series")]
if unsafe { !limbo_series::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_series::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register series extension".to_string());
}
#[cfg(feature = "ipaddr")]
if unsafe { !limbo_ipaddr::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_ipaddr::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register ipaddr extension".to_string());
}
#[cfg(feature = "completion")]
if unsafe { !limbo_completion::register_extension_static(&ext_api).is_ok() } {
if unsafe { !limbo_completion::register_extension_static(&mut ext_api).is_ok() } {
return Err("Failed to register completion extension".to_string());
}
#[cfg(feature = "fs")]
{
let vfslist = add_builtin_vfs_extensions(Some(ext_api)).map_err(|e| e.to_string())?;
for (name, vfs) in vfslist {
add_vfs_module(name, vfs);
}
}
Ok(())
}
}
fn add_vfs_module(name: String, vfs: Arc<VfsMod>) {
let mut modules = VFS_MODULES
.get_or_init(|| Mutex::new(Vec::new()))
.lock()
.unwrap();
if !modules.iter().any(|v| v.0 == name) {
modules.push((name, vfs));
}
}
pub fn list_vfs_modules() -> Vec<String> {
VFS_MODULES
.get_or_init(|| Mutex::new(Vec::new()))
.lock()
.unwrap()
.iter()
.map(|v| v.0.clone())
.collect()
}
fn get_vfs_modules() -> Vec<Vfs> {
VFS_MODULES
.get_or_init(|| Mutex::new(Vec::new()))
.lock()
.unwrap()
.clone()
}

11
core/io/mod.rs
View File

@@ -19,11 +19,21 @@ pub trait File: Send + Sync {
fn size(&self) -> Result<u64>;
}
#[derive(Copy, Clone)]
pub enum OpenFlags {
None,
Create,
}
impl OpenFlags {
pub fn to_flags(&self) -> i32 {
match self {
Self::None => 0,
Self::Create => 1,
}
}
}
pub trait IO: Send + Sync {
fn open_file(&self, path: &str, flags: OpenFlags, direct: bool) -> Result<Arc<dyn File>>;
@@ -203,5 +213,6 @@ cfg_block! {
}
mod memory;
mod vfs;
pub use memory::MemoryIO;
mod common;

153
core/io/vfs.rs Normal file
View File

@@ -0,0 +1,153 @@
use crate::ext::VfsMod;
use crate::{LimboError, Result};
use limbo_ext::{VfsFileImpl, VfsImpl};
use std::cell::RefCell;
use std::ffi::{c_void, CString};
use std::sync::Arc;
use super::{Buffer, Completion, File, OpenFlags, IO};
impl IO for VfsMod {
fn open_file(&self, path: &str, flags: OpenFlags, direct: bool) -> Result<Arc<dyn File>> {
let c_path = CString::new(path).map_err(|_| {
LimboError::ExtensionError("Failed to convert path to CString".to_string())
})?;
let ctx = self.ctx as *mut c_void;
let vfs = unsafe { &*self.ctx };
let file = unsafe { (vfs.open)(ctx, c_path.as_ptr(), flags.to_flags(), direct) };
if file.is_null() {
return Err(LimboError::ExtensionError("File not found".to_string()));
}
Ok(Arc::new(limbo_ext::VfsFileImpl::new(file, self.ctx)?))
}
fn run_once(&self) -> Result<()> {
if self.ctx.is_null() {
return Err(LimboError::ExtensionError("VFS is null".to_string()));
}
let vfs = unsafe { &*self.ctx };
let result = unsafe { (vfs.run_once)(vfs.vfs) };
if !result.is_ok() {
return Err(LimboError::ExtensionError(result.to_string()));
}
Ok(())
}
fn generate_random_number(&self) -> i64 {
if self.ctx.is_null() {
return -1;
}
let vfs = unsafe { &*self.ctx };
unsafe { (vfs.gen_random_number)() }
}
fn get_current_time(&self) -> String {
if self.ctx.is_null() {
return "".to_string();
}
unsafe {
let vfs = &*self.ctx;
let chars = (vfs.current_time)();
let cstr = CString::from_raw(chars as *mut i8);
cstr.to_string_lossy().into_owned()
}
}
}
impl File for VfsFileImpl {
fn lock_file(&self, exclusive: bool) -> Result<()> {
let vfs = unsafe { &*self.vfs };
let result = unsafe { (vfs.lock)(self.file, exclusive) };
if result.is_ok() {
return Err(LimboError::ExtensionError(result.to_string()));
}
Ok(())
}
fn unlock_file(&self) -> Result<()> {
if self.vfs.is_null() {
return Err(LimboError::ExtensionError("VFS is null".to_string()));
}
let vfs = unsafe { &*self.vfs };
let result = unsafe { (vfs.unlock)(self.file) };
if result.is_ok() {
return Err(LimboError::ExtensionError(result.to_string()));
}
Ok(())
}
fn pread(&self, pos: usize, c: Completion) -> Result<()> {
let r = match &c {
Completion::Read(ref r) => r,
_ => unreachable!(),
};
let result = {
let mut buf = r.buf_mut();
let count = buf.len();
let vfs = unsafe { &*self.vfs };
unsafe { (vfs.read)(self.file, buf.as_mut_ptr(), count, pos as i64) }
};
if result < 0 {
Err(LimboError::ExtensionError("pread failed".to_string()))
} else {
c.complete(0);
Ok(())
}
}
fn pwrite(&self, pos: usize, buffer: Arc<RefCell<Buffer>>, c: Completion) -> Result<()> {
let buf = buffer.borrow();
let count = buf.as_slice().len();
if self.vfs.is_null() {
return Err(LimboError::ExtensionError("VFS is null".to_string()));
}
let vfs = unsafe { &*self.vfs };
let result = unsafe {
(vfs.write)(
self.file,
buf.as_slice().as_ptr() as *mut u8,
count,
pos as i64,
)
};
if result < 0 {
Err(LimboError::ExtensionError("pwrite failed".to_string()))
} else {
c.complete(result);
Ok(())
}
}
fn sync(&self, c: Completion) -> Result<()> {
let vfs = unsafe { &*self.vfs };
let result = unsafe { (vfs.sync)(self.file) };
if result < 0 {
Err(LimboError::ExtensionError("sync failed".to_string()))
} else {
c.complete(0);
Ok(())
}
}
fn size(&self) -> Result<u64> {
let vfs = unsafe { &*self.vfs };
let result = unsafe { (vfs.size)(self.file) };
if result < 0 {
Err(LimboError::ExtensionError("size failed".to_string()))
} else {
Ok(result as u64)
}
}
}
impl Drop for VfsMod {
fn drop(&mut self) {
if self.ctx.is_null() {
return;
}
unsafe {
let _ = Box::from_raw(self.ctx as *mut VfsImpl);
}
}
}

101
core/lib.rs
View File

@@ -23,11 +23,8 @@ mod vector;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
use ext::list_vfs_modules;
use fallible_iterator::FallibleIterator;
#[cfg(not(target_family = "wasm"))]
use libloading::{Library, Symbol};
#[cfg(not(target_family = "wasm"))]
use limbo_ext::{ExtensionApi, ExtensionEntryPoint};
use limbo_ext::{ResultCode, VTabKind, VTabModuleImpl};
use limbo_sqlite3_parser::{ast, ast::Cmd, lexer::sql::Parser};
use parking_lot::RwLock;
@@ -204,6 +201,31 @@ impl Database {
}
Ok(conn)
}
/// Open a new database file with a specified VFS without an existing database
/// connection and symbol table to register extensions.
#[cfg(feature = "fs")]
#[allow(clippy::arc_with_non_send_sync)]
pub fn open_new(path: &str, vfs: &str) -> Result<(Arc<dyn IO>, Arc<Database>)> {
let vfsmods = ext::add_builtin_vfs_extensions(None)?;
let io: Arc<dyn IO> = match vfsmods.iter().find(|v| v.0 == vfs).map(|v| v.1.clone()) {
Some(vfs) => vfs,
None => match vfs.trim() {
"memory" => Arc::new(MemoryIO::new()),
"syscall" => Arc::new(PlatformIO::new()?),
#[cfg(all(target_os = "linux", feature = "io_uring"))]
"io_uring" => Arc::new(UringIO::new()?),
other => {
return Err(LimboError::InvalidArgument(format!(
"no such VFS: {}",
other
)));
}
},
};
let db = Self::open_file(io.clone(), path, false)?;
Ok((io, db))
}
}
pub fn maybe_init_database_file(file: &Arc<dyn File>, io: &Arc<dyn IO>) -> Result<()> {
@@ -279,8 +301,7 @@ impl Connection {
match cmd {
Cmd::Stmt(stmt) => {
let program = Rc::new(translate::translate(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -321,8 +342,7 @@ impl Connection {
match cmd {
Cmd::Stmt(stmt) => {
let program = Rc::new(translate::translate(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -338,8 +358,7 @@ impl Connection {
}
Cmd::Explain(stmt) => {
let program = translate::translate(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -357,8 +376,7 @@ impl Connection {
match stmt {
ast::Stmt::Select(select) => {
let mut plan = prepare_select_plan(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -368,8 +386,7 @@ impl Connection {
)?;
optimize_plan(
&mut plan,
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -398,8 +415,7 @@ impl Connection {
match cmd {
Cmd::Explain(stmt) => {
let program = translate::translate(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -415,8 +431,7 @@ impl Connection {
Cmd::ExplainQueryPlan(_stmt) => todo!(),
Cmd::Stmt(stmt) => {
let program = translate::translate(
&self
.schema
self.schema
.try_read()
.ok_or(LimboError::SchemaLocked)?
.deref(),
@@ -461,30 +476,6 @@ impl Connection {
Ok(checkpoint_result)
}
#[cfg(not(target_family = "wasm"))]
pub fn load_extension<P: AsRef<std::ffi::OsStr>>(&self, path: P) -> Result<()> {
let api = Box::new(self.build_limbo_ext());
let lib =
unsafe { Library::new(path).map_err(|e| LimboError::ExtensionError(e.to_string()))? };
let entry: Symbol<ExtensionEntryPoint> = unsafe {
lib.get(b"register_extension")
.map_err(|e| LimboError::ExtensionError(e.to_string()))?
};
let api_ptr: *const ExtensionApi = Box::into_raw(api);
let result_code = unsafe { entry(api_ptr) };
if result_code.is_ok() {
self.syms.borrow_mut().extensions.push((lib, api_ptr));
Ok(())
} else {
if !api_ptr.is_null() {
let _ = unsafe { Box::from_raw(api_ptr.cast_mut()) };
}
Err(LimboError::ExtensionError(
"Extension registration failed".to_string(),
))
}
}
/// Close a connection and checkpoint.
pub fn close(&self) -> Result<()> {
loop {
@@ -517,6 +508,28 @@ impl Connection {
pub fn total_changes(&self) -> i64 {
self.total_changes.get()
}
#[cfg(feature = "fs")]
pub fn open_new(&self, path: &str, vfs: &str) -> Result<(Arc<dyn IO>, Arc<Database>)> {
Database::open_with_vfs(&self._db, path, vfs)
}
pub fn list_vfs(&self) -> Vec<String> {
let mut all_vfs = vec![String::from("memory")];
#[cfg(feature = "fs")]
{
#[cfg(all(feature = "fs", target_family = "unix"))]
{
all_vfs.push("syscall".to_string());
}
#[cfg(all(feature = "fs", target_os = "linux", feature = "io_uring"))]
{
all_vfs.push("io_uring".to_string());
}
}
all_vfs.extend(list_vfs_modules());
all_vfs
}
}
pub struct Statement {
@@ -723,8 +736,6 @@ impl VirtualTable {
pub(crate) struct SymbolTable {
pub functions: HashMap<String, Rc<function::ExternalFunc>>,
#[cfg(not(target_family = "wasm"))]
extensions: Vec<(Library, *const ExtensionApi)>,
pub vtabs: HashMap<String, Rc<VirtualTable>>,
pub vtab_modules: HashMap<String, Rc<crate::ext::VTabImpl>>,
}
@@ -769,8 +780,6 @@ impl SymbolTable {
Self {
functions: HashMap::new(),
vtabs: HashMap::new(),
#[cfg(not(target_family = "wasm"))]
extensions: Vec::new(),
vtab_modules: HashMap::new(),
}
}

4
extensions/core/Cargo.toml
View File

@@ -13,3 +13,7 @@ static = []
[dependencies]
limbo_macros = { workspace = true }
[target.'cfg(not(target_family = "wasm"))'.dependencies]
getrandom = "0.3.1"
chrono = "0.4.40"

107
extensions/core/README.md
View File

@@ -10,7 +10,7 @@ like traditional `sqlite3` extensions, but are able to be written in much more e
- [ x ] **Scalar Functions**: Create scalar functions using the `scalar` macro.
- [ x ] **Aggregate Functions**: Define aggregate functions with `AggregateDerive` macro and `AggFunc` trait.
- [ x ] **Virtual tables**: Create a module for a virtual table with the `VTabModuleDerive` macro and `VTabCursor` trait.
- [] **VFS Modules**
- [ x ] **VFS Modules**: Extend Limbo's OS interface by implementing `VfsExtension` and `VfsFile` traits.
---
## Installation
@@ -59,9 +59,14 @@ register_extension!{
scalars: { double }, // name of your function, if different from attribute name
aggregates: { Percentile },
vtabs: { CsvVTable },
vfs: { ExampleFS },
}
```
**NOTE**: Currently, any Derive macro used from this crate is required to be in the same
file as the `register_extension` macro.
### Scalar Example:
```rust
use limbo_ext::{register_extension, Value, scalar};
@@ -279,6 +284,106 @@ impl VTabCursor for CsvCursor {
}
```
### VFS Example
```rust
use limbo_ext::{ExtResult as Result, VfsDerive, VfsExtension, VfsFile};
/// Your struct must also impl Default
#[derive(VfsDerive, Default)]
struct ExampleFS;
struct ExampleFile {
file: std::fs::File,
}
impl VfsExtension for ExampleFS {
/// The name of your vfs module
const NAME: &'static str = "example";
type File = ExampleFile;
fn open(&self, path: &str, flags: i32, _direct: bool) -> Result<Self::File> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(flags & 1 != 0)
.open(path)
.map_err(|_| ResultCode::Error)?;
Ok(TestFile { file })
}
fn run_once(&self) -> Result<()> {
// (optional) method to cycle/advance IO, if your extension is asynchronous
Ok(())
}
fn close(&self, file: Self::File) -> Result<()> {
// (optional) method to close or drop the file
Ok(())
}
fn generate_random_number(&self) -> i64 {
// (optional) method to generate random number. Used for testing
let mut buf = [0u8; 8];
getrandom::fill(&mut buf).unwrap();
i64::from_ne_bytes(buf)
}
fn get_current_time(&self) -> String {
// (optional) method to generate random number. Used for testing
chrono::Local::now().format("%Y-%m-%d %H:%M:%S").to_string()
}
}
impl VfsFile for ExampleFile {
fn read(
&mut self,
buf: &mut [u8],
count: usize,
offset: i64,
) -> Result<i32> {
if file.file.seek(SeekFrom::Start(offset as u64)).is_err() {
return Err(ResultCode::Error);
}
file.file
.read(&mut buf[..count])
.map_err(|_| ResultCode::Error)
.map(|n| n as i32)
}
fn write(&mut self, buf: &[u8], count: usize, offset: i64) -> Result<i32> {
if self.file.seek(SeekFrom::Start(offset as u64)).is_err() {
return Err(ResultCode::Error);
}
self.file
.write(&buf[..count])
.map_err(|_| ResultCode::Error)
.map(|n| n as i32)
}
fn sync(&self) -> Result<()> {
self.file.sync_all().map_err(|_| ResultCode::Error)
}
fn lock(&self, _exclusive: bool) -> Result<()> {
// (optional) method to lock the file
Ok(())
}
fn unlock(&self) -> Result<()> {
// (optional) method to lock the file
Ok(())
}
fn size(&self) -> i64 {
self.file.metadata().map(|m| m.len() as i64).unwrap_or(-1)
}
}
```
## Cargo.toml Config
Edit the workspace `Cargo.toml` to include your extension as a workspace dependency, e.g:

36
extensions/core/src/lib.rs
View File

@@ -1,10 +1,16 @@
mod types;
mod vfs_modules;
#[cfg(not(target_family = "wasm"))]
pub use limbo_macros::VfsDerive;
pub use limbo_macros::{register_extension, scalar, AggregateDerive, VTabModuleDerive};
use std::{
fmt::Display,
os::raw::{c_char, c_void},
};
pub use types::{ResultCode, Value, ValueType};
pub use vfs_modules::{RegisterVfsFn, VfsFileImpl, VfsImpl};
#[cfg(not(target_family = "wasm"))]
pub use vfs_modules::{VfsExtension, VfsFile};
pub type ExtResult<T> = std::result::Result<T, ResultCode>;
@@ -14,6 +20,36 @@ pub struct ExtensionApi {
pub register_scalar_function: RegisterScalarFn,
pub register_aggregate_function: RegisterAggFn,
pub register_module: RegisterModuleFn,
pub register_vfs: RegisterVfsFn,
pub builtin_vfs: *mut *const VfsImpl,
pub builtin_vfs_count: i32,
}
unsafe impl Send for ExtensionApi {}
unsafe impl Send for ExtensionApiRef {}
#[repr(C)]
pub struct ExtensionApiRef {
pub api: *const ExtensionApi,
}
impl ExtensionApi {
/// Since we want the option to build in extensions at compile time as well,
/// we add a slice of VfsImpls to the extension API, and this is called with any
/// libraries that we load staticly that will add their VFS implementations to the list.
pub fn add_builtin_vfs(&mut self, vfs: *const VfsImpl) -> ResultCode {
if vfs.is_null() || self.builtin_vfs.is_null() {
return ResultCode::Error;
}
let mut new = unsafe {
let slice =
std::slice::from_raw_parts_mut(self.builtin_vfs, self.builtin_vfs_count as usize);
Vec::from(slice)
};
new.push(vfs);
self.builtin_vfs = Box::into_raw(new.into_boxed_slice()) as *mut *const VfsImpl;
self.builtin_vfs_count += 1;
ResultCode::OK
}
}
pub type ExtensionEntryPoint = unsafe extern "C" fn(api: *const ExtensionApi) -> ResultCode;

3
extensions/core/src/types.rs
View File

@@ -165,6 +165,7 @@ impl TextValue {
})
}
#[cfg(feature = "core_only")]
fn free(self) {
if !self.text.is_null() {
let _ = unsafe { Box::from_raw(self.text as *mut u8) };
@@ -231,7 +232,7 @@ impl Blob {
}
unsafe { std::slice::from_raw_parts(self.data, self.size as usize) }
}
#[cfg(feature = "core_only")]
fn free(self) {
if !self.data.is_null() {
let _ = unsafe { Box::from_raw(self.data as *mut u8) };

114
extensions/core/src/vfs_modules.rs Normal file
View File

@@ -0,0 +1,114 @@
use crate::{ExtResult, ResultCode};
use std::ffi::{c_char, c_void};
#[cfg(not(target_family = "wasm"))]
pub trait VfsExtension: Default + Send + Sync {
const NAME: &'static str;
type File: VfsFile;
fn open_file(&self, path: &str, flags: i32, direct: bool) -> ExtResult<Self::File>;
fn run_once(&self) -> ExtResult<()> {
Ok(())
}
fn close(&self, _file: Self::File) -> ExtResult<()> {
Ok(())
}
fn generate_random_number(&self) -> i64 {
let mut buf = [0u8; 8];
getrandom::fill(&mut buf).unwrap();
i64::from_ne_bytes(buf)
}
fn get_current_time(&self) -> String {
chrono::Local::now().format("%Y-%m-%d %H:%M:%S").to_string()
}
}
#[cfg(not(target_family = "wasm"))]
pub trait VfsFile: Send + Sync {
fn lock(&mut self, _exclusive: bool) -> ExtResult<()> {
Ok(())
}
fn unlock(&self) -> ExtResult<()> {
Ok(())
}
fn read(&mut self, buf: &mut [u8], count: usize, offset: i64) -> ExtResult<i32>;
fn write(&mut self, buf: &[u8], count: usize, offset: i64) -> ExtResult<i32>;
fn sync(&self) -> ExtResult<()>;
fn size(&self) -> i64;
}
#[repr(C)]
pub struct VfsImpl {
pub name: *const c_char,
pub vfs: *const c_void,
pub open: VfsOpen,
pub close: VfsClose,
pub read: VfsRead,
pub write: VfsWrite,
pub sync: VfsSync,
pub lock: VfsLock,
pub unlock: VfsUnlock,
pub size: VfsSize,
pub run_once: VfsRunOnce,
pub current_time: VfsGetCurrentTime,
pub gen_random_number: VfsGenerateRandomNumber,
}
pub type RegisterVfsFn =
unsafe extern "C" fn(name: *const c_char, vfs: *const VfsImpl) -> ResultCode;
pub type VfsOpen = unsafe extern "C" fn(
ctx: *const c_void,
path: *const c_char,
flags: i32,
direct: bool,
) -> *const c_void;
pub type VfsClose = unsafe extern "C" fn(file: *const c_void) -> ResultCode;
pub type VfsRead =
unsafe extern "C" fn(file: *const c_void, buf: *mut u8, count: usize, offset: i64) -> i32;
pub type VfsWrite =
unsafe extern "C" fn(file: *const c_void, buf: *const u8, count: usize, offset: i64) -> i32;
pub type VfsSync = unsafe extern "C" fn(file: *const c_void) -> i32;
pub type VfsLock = unsafe extern "C" fn(file: *const c_void, exclusive: bool) -> ResultCode;
pub type VfsUnlock = unsafe extern "C" fn(file: *const c_void) -> ResultCode;
pub type VfsSize = unsafe extern "C" fn(file: *const c_void) -> i64;
pub type VfsRunOnce = unsafe extern "C" fn(file: *const c_void) -> ResultCode;
pub type VfsGetCurrentTime = unsafe extern "C" fn() -> *const c_char;
pub type VfsGenerateRandomNumber = unsafe extern "C" fn() -> i64;
#[repr(C)]
pub struct VfsFileImpl {
pub file: *const c_void,
pub vfs: *const VfsImpl,
}
unsafe impl Send for VfsFileImpl {}
unsafe impl Sync for VfsFileImpl {}
impl VfsFileImpl {
pub fn new(file: *const c_void, vfs: *const VfsImpl) -> ExtResult<Self> {
if file.is_null() || vfs.is_null() {
return Err(ResultCode::Error);
}
Ok(Self { file, vfs })
}
}
impl Drop for VfsFileImpl {
fn drop(&mut self) {
if self.vfs.is_null() || self.file.is_null() {
return;
}
let vfs = unsafe { &*self.vfs };
unsafe {
(vfs.close)(self.file);
}
}
}

6
extensions/kvstore/Cargo.toml → extensions/tests/Cargo.toml
View File

@@ -1,5 +1,5 @@
[package]
name = "limbo_kv"
name = "limbo_ext_tests"
version.workspace = true
authors.workspace = true
edition.workspace = true
@@ -13,8 +13,10 @@ crate-type = ["cdylib", "lib"]
static= [ "limbo_ext/static" ]
[dependencies]
env_logger = "0.11.6"
lazy_static = "1.5.0"
limbo_ext = { workspace = true, features = ["static"] }
log = "0.4.26"
[target.'cfg(not(target_family = "wasm"))'.dependencies]
mimalloc = { version = "*", default-features = false }
mimalloc = { version = "0.1", default-features = false }

88
extensions/kvstore/src/lib.rs → extensions/tests/src/lib.rs
View File

@@ -1,16 +1,23 @@
use lazy_static::lazy_static;
use limbo_ext::{
register_extension, ResultCode, VTabCursor, VTabKind, VTabModule, VTabModuleDerive, Value,
register_extension, scalar, ExtResult, ResultCode, VTabCursor, VTabKind, VTabModule,
VTabModuleDerive, Value,
};
#[cfg(not(target_family = "wasm"))]
use limbo_ext::{VfsDerive, VfsExtension, VfsFile};
use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::sync::Mutex;
lazy_static! {
static ref GLOBAL_STORE: Mutex<BTreeMap<i64, (String, String)>> = Mutex::new(BTreeMap::new());
}
register_extension! {
vtabs: { KVStoreVTab },
scalars: { test_scalar },
vfs: { TestFS },
}
lazy_static! {
static ref GLOBAL_STORE: Mutex<BTreeMap<i64, (String, String)>> = Mutex::new(BTreeMap::new());
}
#[derive(VTabModuleDerive, Default)]
@@ -128,7 +135,7 @@ impl VTabCursor for KVStoreCursor {
if self.index.is_some_and(|c| c < self.rows.len()) {
self.rows[self.index.unwrap_or(0)].0
} else {
println!("rowid: -1");
log::error!("rowid: -1");
-1
}
}
@@ -145,3 +152,72 @@ impl VTabCursor for KVStoreCursor {
<KVStoreVTab as VTabModule>::next(self)
}
}
pub struct TestFile {
file: File,
}
#[cfg(target_family = "wasm")]
pub struct TestFS;
#[cfg(not(target_family = "wasm"))]
#[derive(VfsDerive, Default)]
pub struct TestFS;
// Test that we can have additional extension types in the same file
// and still register the vfs at comptime if linking staticly
#[scalar(name = "test_scalar")]
fn test_scalar(_args: limbo_ext::Value) -> limbo_ext::Value {
limbo_ext::Value::from_integer(42)
}
#[cfg(not(target_family = "wasm"))]
impl VfsExtension for TestFS {
const NAME: &'static str = "testvfs";
type File = TestFile;
fn open_file(&self, path: &str, flags: i32, _direct: bool) -> ExtResult<Self::File> {
let _ = env_logger::try_init();
log::debug!("opening file with testing VFS: {} flags: {}", path, flags);
let file = OpenOptions::new()
.read(true)
.write(true)
.create(flags & 1 != 0)
.open(path)
.map_err(|_| ResultCode::Error)?;
Ok(TestFile { file })
}
}
#[cfg(not(target_family = "wasm"))]
impl VfsFile for TestFile {
fn read(&mut self, buf: &mut [u8], count: usize, offset: i64) -> ExtResult<i32> {
log::debug!("reading file with testing VFS: bytes: {count} offset: {offset}");
if self.file.seek(SeekFrom::Start(offset as u64)).is_err() {
return Err(ResultCode::Error);
}
self.file
.read(&mut buf[..count])
.map_err(|_| ResultCode::Error)
.map(|n| n as i32)
}
fn write(&mut self, buf: &[u8], count: usize, offset: i64) -> ExtResult<i32> {
log::debug!("writing to file with testing VFS: bytes: {count} offset: {offset}");
if self.file.seek(SeekFrom::Start(offset as u64)).is_err() {
return Err(ResultCode::Error);
}
self.file
.write(&buf[..count])
.map_err(|_| ResultCode::Error)
.map(|n| n as i32)
}
fn sync(&self) -> ExtResult<()> {
log::debug!("syncing file with testing VFS");
self.file.sync_all().map_err(|_| ResultCode::Error)
}
fn size(&self) -> i64 {
self.file.metadata().map(|m| m.len() as i64).unwrap_or(-1)
}
}

6
macros/src/args.rs
View File

@@ -7,6 +7,7 @@ pub(crate) struct RegisterExtensionInput {
pub aggregates: Vec<Ident>,
pub scalars: Vec<Ident>,
pub vtabs: Vec<Ident>,
pub vfs: Vec<Ident>,
}
impl syn::parse::Parse for RegisterExtensionInput {
@@ -14,11 +15,12 @@ impl syn::parse::Parse for RegisterExtensionInput {
let mut aggregates = Vec::new();
let mut scalars = Vec::new();
let mut vtabs = Vec::new();
let mut vfs = Vec::new();
while !input.is_empty() {
if input.peek(syn::Ident) && input.peek2(Token![:]) {
let section_name: Ident = input.parse()?;
input.parse::<Token![:]>()?;
let names = ["aggregates", "scalars", "vtabs"];
let names = ["aggregates", "scalars", "vtabs", "vfs"];
if names.contains(&section_name.to_string().as_str()) {
let content;
syn::braced!(content in input);
@@ -30,6 +32,7 @@ impl syn::parse::Parse for RegisterExtensionInput {
"aggregates" => aggregates = parsed_items,
"scalars" => scalars = parsed_items,
"vtabs" => vtabs = parsed_items,
"vfs" => vfs = parsed_items,
_ => unreachable!(),
};
@@ -48,6 +51,7 @@ impl syn::parse::Parse for RegisterExtensionInput {
aggregates,
scalars,
vtabs,
vfs,
})
}
}

249
macros/src/lib.rs
View File

@@ -623,6 +623,222 @@ pub fn derive_vtab_module(input: TokenStream) -> TokenStream {
TokenStream::from(expanded)
}
#[proc_macro_derive(VfsDerive)]
pub fn derive_vfs_module(input: TokenStream) -> TokenStream {
let derive_input = parse_macro_input!(input as DeriveInput);
let struct_name = &derive_input.ident;
let register_fn_name = format_ident!("register_{}", struct_name);
let register_static = format_ident!("register_static_{}", struct_name);
let open_fn_name = format_ident!("{}_open", struct_name);
let close_fn_name = format_ident!("{}_close", struct_name);
let read_fn_name = format_ident!("{}_read", struct_name);
let write_fn_name = format_ident!("{}_write", struct_name);
let lock_fn_name = format_ident!("{}_lock", struct_name);
let unlock_fn_name = format_ident!("{}_unlock", struct_name);
let sync_fn_name = format_ident!("{}_sync", struct_name);
let size_fn_name = format_ident!("{}_size", struct_name);
let run_once_fn_name = format_ident!("{}_run_once", struct_name);
let generate_random_number_fn_name = format_ident!("{}_generate_random_number", struct_name);
let get_current_time_fn_name = format_ident!("{}_get_current_time", struct_name);
let expanded = quote! {
#[allow(non_snake_case)]
pub unsafe extern "C" fn #register_static() -> *const ::limbo_ext::VfsImpl {
let ctx = #struct_name::default();
let ctx = ::std::boxed::Box::into_raw(::std::boxed::Box::new(ctx)) as *const ::std::ffi::c_void;
let name = ::std::ffi::CString::new(<#struct_name as ::limbo_ext::VfsExtension>::NAME).unwrap().into_raw();
let vfs_mod = ::limbo_ext::VfsImpl {
vfs: ctx,
name,
open: #open_fn_name,
close: #close_fn_name,
read: #read_fn_name,
write: #write_fn_name,
lock: #lock_fn_name,
unlock: #unlock_fn_name,
sync: #sync_fn_name,
size: #size_fn_name,
run_once: #run_once_fn_name,
gen_random_number: #generate_random_number_fn_name,
current_time: #get_current_time_fn_name,
};
::std::boxed::Box::into_raw(::std::boxed::Box::new(vfs_mod)) as *const ::limbo_ext::VfsImpl
}
#[no_mangle]
pub unsafe extern "C" fn #register_fn_name(api: &::limbo_ext::ExtensionApi) -> ::limbo_ext::ResultCode {
let ctx = #struct_name::default();
let ctx = ::std::boxed::Box::into_raw(::std::boxed::Box::new(ctx)) as *const ::std::ffi::c_void;
let name = ::std::ffi::CString::new(<#struct_name as ::limbo_ext::VfsExtension>::NAME).unwrap().into_raw();
let vfs_mod = ::limbo_ext::VfsImpl {
vfs: ctx,
name,
open: #open_fn_name,
close: #close_fn_name,
read: #read_fn_name,
write: #write_fn_name,
lock: #lock_fn_name,
unlock: #unlock_fn_name,
sync: #sync_fn_name,
size: #size_fn_name,
run_once: #run_once_fn_name,
gen_random_number: #generate_random_number_fn_name,
current_time: #get_current_time_fn_name,
};
let vfsimpl = ::std::boxed::Box::into_raw(::std::boxed::Box::new(vfs_mod)) as *const ::limbo_ext::VfsImpl;
(api.register_vfs)(name, vfsimpl)
}
#[no_mangle]
pub unsafe extern "C" fn #open_fn_name(
ctx: *const ::std::ffi::c_void,
path: *const ::std::ffi::c_char,
flags: i32,
direct: bool,
) -> *const ::std::ffi::c_void {
let ctx = &*(ctx as *const ::limbo_ext::VfsImpl);
let Ok(path_str) = ::std::ffi::CStr::from_ptr(path).to_str() else {
return ::std::ptr::null_mut();
};
let vfs = &*(ctx.vfs as *const #struct_name);
let Ok(file_handle) = <#struct_name as ::limbo_ext::VfsExtension>::open_file(vfs, path_str, flags, direct) else {
return ::std::ptr::null();
};
let boxed = ::std::boxed::Box::into_raw(::std::boxed::Box::new(file_handle)) as *const ::std::ffi::c_void;
let Ok(vfs_file) = ::limbo_ext::VfsFileImpl::new(boxed, ctx) else {
return ::std::ptr::null();
};
::std::boxed::Box::into_raw(::std::boxed::Box::new(vfs_file)) as *const ::std::ffi::c_void
}
#[no_mangle]
pub unsafe extern "C" fn #close_fn_name(file_ptr: *const ::std::ffi::c_void) -> ::limbo_ext::ResultCode {
if file_ptr.is_null() {
return ::limbo_ext::ResultCode::Error;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let vfs_instance = &*(vfs_file.vfs as *const #struct_name);
// this time we need to own it so we can drop it
let file: ::std::boxed::Box<<#struct_name as ::limbo_ext::VfsExtension>::File> =
::std::boxed::Box::from_raw(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
if let Err(e) = <#struct_name as ::limbo_ext::VfsExtension>::close(vfs_instance, *file) {
return e;
}
::limbo_ext::ResultCode::OK
}
#[no_mangle]
pub unsafe extern "C" fn #read_fn_name(file_ptr: *const ::std::ffi::c_void, buf: *mut u8, count: usize, offset: i64) -> i32 {
if file_ptr.is_null() {
return -1;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
match <#struct_name as ::limbo_ext::VfsExtension>::File::read(file, ::std::slice::from_raw_parts_mut(buf, count), count, offset) {
Ok(n) => n,
Err(_) => -1,
}
}
#[no_mangle]
pub unsafe extern "C" fn #run_once_fn_name(ctx: *const ::std::ffi::c_void) -> ::limbo_ext::ResultCode {
if ctx.is_null() {
return ::limbo_ext::ResultCode::Error;
}
let ctx = &mut *(ctx as *mut #struct_name);
if let Err(e) = <#struct_name as ::limbo_ext::VfsExtension>::run_once(ctx) {
return e;
}
::limbo_ext::ResultCode::OK
}
#[no_mangle]
pub unsafe extern "C" fn #write_fn_name(file_ptr: *const ::std::ffi::c_void, buf: *const u8, count: usize, offset: i64) -> i32 {
if file_ptr.is_null() {
return -1;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
match <#struct_name as ::limbo_ext::VfsExtension>::File::write(file, ::std::slice::from_raw_parts(buf, count), count, offset) {
Ok(n) => n,
Err(_) => -1,
}
}
#[no_mangle]
pub unsafe extern "C" fn #lock_fn_name(file_ptr: *const ::std::ffi::c_void, exclusive: bool) -> ::limbo_ext::ResultCode {
if file_ptr.is_null() {
return ::limbo_ext::ResultCode::Error;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
if let Err(e) = <#struct_name as ::limbo_ext::VfsExtension>::File::lock(file, exclusive) {
return e;
}
::limbo_ext::ResultCode::OK
}
#[no_mangle]
pub unsafe extern "C" fn #unlock_fn_name(file_ptr: *const ::std::ffi::c_void) -> ::limbo_ext::ResultCode {
if file_ptr.is_null() {
return ::limbo_ext::ResultCode::Error;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
if let Err(e) = <#struct_name as ::limbo_ext::VfsExtension>::File::unlock(file) {
return e;
}
::limbo_ext::ResultCode::OK
}
#[no_mangle]
pub unsafe extern "C" fn #sync_fn_name(file_ptr: *const ::std::ffi::c_void) -> i32 {
if file_ptr.is_null() {
return -1;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
if <#struct_name as ::limbo_ext::VfsExtension>::File::sync(file).is_err() {
return -1;
}
0
}
#[no_mangle]
pub unsafe extern "C" fn #size_fn_name(file_ptr: *const ::std::ffi::c_void) -> i64 {
if file_ptr.is_null() {
return -1;
}
let vfs_file: &mut ::limbo_ext::VfsFileImpl = &mut *(file_ptr as *mut ::limbo_ext::VfsFileImpl);
let file: &mut <#struct_name as ::limbo_ext::VfsExtension>::File =
&mut *(vfs_file.file as *mut <#struct_name as ::limbo_ext::VfsExtension>::File);
<#struct_name as ::limbo_ext::VfsExtension>::File::size(file)
}
#[no_mangle]
pub unsafe extern "C" fn #generate_random_number_fn_name() -> i64 {
let obj = #struct_name::default();
<#struct_name as ::limbo_ext::VfsExtension>::generate_random_number(&obj)
}
#[no_mangle]
pub unsafe extern "C" fn #get_current_time_fn_name() -> *const ::std::ffi::c_char {
let obj = #struct_name::default();
let time = <#struct_name as ::limbo_ext::VfsExtension>::get_current_time(&obj);
// release ownership of the string to core
::std::ffi::CString::new(time).unwrap().into_raw() as *const ::std::ffi::c_char
}
};
TokenStream::from(expanded)
}
/// Register your extension with 'core' by providing the relevant functions
///```ignore
///use limbo_ext::{register_extension, scalar, Value, AggregateDerive, AggFunc};
@@ -662,6 +878,7 @@ pub fn register_extension(input: TokenStream) -> TokenStream {
aggregates,
scalars,
vtabs,
vfs,
} = input_ast;
let scalar_calls = scalars.iter().map(|scalar_ident| {
@@ -699,6 +916,29 @@ pub fn register_extension(input: TokenStream) -> TokenStream {
}
}
});
let vfs_calls = vfs.iter().map(|vfs_ident| {
let register_fn = syn::Ident::new(&format!("register_{}", vfs_ident), vfs_ident.span());
quote! {
{
let result = unsafe { #register_fn(api) };
if !result.is_ok() {
return result;
}
}
}
});
let static_vfs = vfs.iter().map(|vfs_ident| {
let static_register =
syn::Ident::new(&format!("register_static_{}", vfs_ident), vfs_ident.span());
quote! {
{
let result = api.add_builtin_vfs(unsafe { #static_register()});
if !result.is_ok() {
return result;
}
}
}
});
let static_aggregates = aggregate_calls.clone();
let static_scalars = scalar_calls.clone();
let static_vtabs = vtab_calls.clone();
@@ -710,27 +950,30 @@ pub fn register_extension(input: TokenStream) -> TokenStream {
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
#[cfg(feature = "static")]
pub unsafe extern "C" fn register_extension_static(api: &::limbo_ext::ExtensionApi) -> ::limbo_ext::ResultCode {
let api = unsafe { &*api };
pub unsafe extern "C" fn register_extension_static(api: &mut ::limbo_ext::ExtensionApi) -> ::limbo_ext::ResultCode {
#(#static_scalars)*
#(#static_aggregates)*
#(#static_vtabs)*
#[cfg(not(target_family = "wasm"))]
#(#static_vfs)*
::limbo_ext::ResultCode::OK
}
#[cfg(not(feature = "static"))]
#[no_mangle]
pub unsafe extern "C" fn register_extension(api: &::limbo_ext::ExtensionApi) -> ::limbo_ext::ResultCode {
let api = unsafe { &*api };
#(#scalar_calls)*
#(#aggregate_calls)*
#(#vtab_calls)*
#(#vfs_calls)*
::limbo_ext::ResultCode::OK
}
};

84
testing/cli_tests/extensions.py
View File

@@ -337,7 +337,7 @@ def test_series():
def test_kv():
ext_path = "./target/debug/liblimbo_kv"
ext_path = "target/debug/liblimbo_ext_tests"
limbo = TestLimboShell()
limbo.run_test_fn(
"create virtual table t using kv_store;",
@@ -401,17 +401,18 @@ def test_kv():
)
limbo.quit()
def test_ipaddr():
limbo = TestLimboShell()
ext_path = "./target/debug/liblimbo_ipaddr"
limbo.run_test_fn(
"SELECT ipfamily('192.168.1.1');",
lambda res: "error: no such function: " in res,
"ipfamily function returns null when ext not loaded",
)
limbo.execute_dot(f".load {ext_path}")
limbo.run_test_fn(
"SELECT ipfamily('192.168.1.1');",
lambda res: "4" == res,
@@ -455,7 +456,7 @@ def test_ipaddr():
lambda res: "128" == res,
"ipmasklen function returns the mask length for IPv6",
)
limbo.run_test_fn(
"SELECT ipnetwork('192.168.16.12/24');",
lambda res: "192.168.16.0/24" == res,
@@ -466,7 +467,76 @@ def test_ipaddr():
lambda res: "2001:db8::1/128" == res,
"ipnetwork function returns the network for IPv6",
)
limbo.quit()
def test_vfs():
limbo = TestLimboShell()
ext_path = "target/debug/liblimbo_ext_tests"
limbo.run_test_fn(".vfslist", lambda x: "testvfs" not in x, "testvfs not loaded")
limbo.execute_dot(f".load {ext_path}")
limbo.run_test_fn(
".vfslist", lambda res: "testvfs" in res, "testvfs extension loaded"
)
limbo.execute_dot(".open testing/vfs.db testvfs")
limbo.execute_dot("create table test (id integer primary key, value float);")
limbo.execute_dot("create table vfs (id integer primary key, value blob);")
for i in range(50):
limbo.execute_dot("insert into test (value) values (randomblob(32*1024));")
limbo.execute_dot(f"insert into vfs (value) values ({i});")
limbo.run_test_fn(
"SELECT count(*) FROM test;",
lambda res: res == "50",
"Tested large write to testfs",
)
limbo.run_test_fn(
"SELECT count(*) FROM vfs;",
lambda res: res == "50",
"Tested large write to testfs",
)
print("Tested large write to testfs")
# open regular db file to ensure we don't segfault when vfs file is dropped
limbo.execute_dot(".open testing/vfs.db")
limbo.execute_dot("create table test (id integer primary key, value float);")
limbo.execute_dot("insert into test (value) values (1.0);")
limbo.quit()
def test_sqlite_vfs_compat():
sqlite = TestLimboShell(
init_commands="",
exec_name="sqlite3",
flags="testing/vfs.db",
)
sqlite.run_test_fn(
".show",
lambda res: "filename: testing/vfs.db" in res,
"Opened db file created with vfs extension in sqlite3",
)
sqlite.run_test_fn(
".schema",
lambda res: "CREATE TABLE test (id integer PRIMARY KEY, value float);" in res,
"Tables created by vfs extension exist in db file",
)
sqlite.run_test_fn(
"SELECT count(*) FROM test;",
lambda res: res == "50",
"Tested large write to testfs",
)
sqlite.run_test_fn(
"SELECT count(*) FROM vfs;",
lambda res: res == "50",
"Tested large write to testfs",
)
sqlite.quit()
def cleanup():
if os.path.exists("testing/vfs.db"):
os.remove("testing/vfs.db")
if os.path.exists("testing/vfs.db-wal"):
os.remove("testing/vfs.db-wal")
if __name__ == "__main__":
try:
@@ -477,7 +547,11 @@ if __name__ == "__main__":
test_series()
test_kv()
test_ipaddr()
test_vfs()
test_sqlite_vfs_compat()
except Exception as e:
print(f"Test FAILED: {e}")
cleanup()
exit(1)
cleanup()
print("All tests passed successfully.")

37
testing/cli_tests/test_limbo_cli.py
View File

@@ -1,8 +1,8 @@
#!/usr/bin/env python3
import os
import select
from time import sleep
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, List, Optional
@@ -10,16 +10,14 @@ from typing import Callable, List, Optional
PIPE_BUF = 4096
@dataclass
class ShellConfig:
sqlite_exec: str = os.getenv("LIMBO_TARGET", "./target/debug/limbo")
sqlite_flags: List[str] = field(
default_factory=lambda: os.getenv("SQLITE_FLAGS", "-q").split()
)
cwd = os.getcwd()
test_dir: Path = field(default_factory=lambda: Path("testing"))
py_folder: Path = field(default_factory=lambda: Path("cli_tests"))
test_files: Path = field(default_factory=lambda: Path("test_files"))
def __init__(self, exe_name, flags: str = "-q"):
self.sqlite_exec: str = exe_name
self.sqlite_flags: List[str] = flags.split()
self.cwd = os.getcwd()
self.test_dir: Path = Path("testing")
self.py_folder: Path = Path("cli_tests")
self.test_files: Path = Path("test_files")
class LimboShell:
@@ -92,14 +90,24 @@ class LimboShell:
def quit(self) -> None:
self._write_to_pipe(".quit")
sleep(0.3)
self.pipe.terminate()
self.pipe.kill()
class TestLimboShell:
def __init__(
self, init_commands: Optional[str] = None, init_blobs_table: bool = False
self,
init_commands: Optional[str] = None,
init_blobs_table: bool = False,
exec_name: Optional[str] = None,
flags="",
):
self.config = ShellConfig()
if exec_name is None:
exec_name = "./target/debug/limbo"
if flags == "":
flags = "-q"
self.config = ShellConfig(exe_name=exec_name, flags=flags)
if init_commands is None:
# Default initialization
init_commands = """
@@ -132,6 +140,11 @@ INSERT INTO t VALUES (zeroblob(1024 - 1), zeroblob(1024 - 2), zeroblob(1024 - 3)
f"Actual:\n{repr(actual)}"
)
def debug_print(self, sql: str):
print(f"debugging: {sql}")
actual = self.shell.execute(sql)
print(f"OUTPUT:\n{repr(actual)}")
def run_test_fn(
self, sql: str, validate: Callable[[str], bool], desc: str = ""
) -> None: