feat(rust): introduce a plug-in mechanism for rust modules

rust/lance/Cargo.toml

@@ -76,6 +76,7 @@ tempfile.workspace = true
 tracing.workspace = true
 lazy_static = { workspace = true }
 async_cell = "0.2.2"
+libloading = "0.8.6"
 
 [target.'cfg(target_os = "linux")'.dev-dependencies]
 pprof.workspace = true
@@ -98,6 +99,7 @@ env_logger = "0.10.0"
 tracing-chrome = "0.7.1"
 rstest = { workspace = true }
 random_word = { version = "0.4.3", features = ["en"] }
+libloading = { version = "0.8.6"}
 
 
 [features]
@@ -114,6 +116,7 @@ protoc = [
     "lance-index/protoc",
     "lance-table/protoc",
 ]
+test-plugin = []
 
 [[bin]]
 name = "lq"
@@ -141,3 +144,10 @@ harness = false
 
 [lints]
 workspace = true
+
+[[example]]
+name = "test_plugin"
+crate-type = ["cdylib"]  # 必须指定为cdylib
+path = "src/plugin/test_plugin.rs"
+required-features = ["test-plugin"]  # 绑定到特性开关
+

rust/lance/src/lib.rs

@@ -82,6 +82,7 @@ pub mod io;
 pub mod session;
 pub mod table;
 pub mod utils;
+pub mod plugin;
 
 pub use dataset::Dataset;
 use lance_index::vector::DIST_COL;

rust/lance/src/plugin.rs

@@ -0,0 +1,272 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use libloading::{Library, Symbol};
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use serde_json::Value;
+use std::fmt;
+
+const CURRENT_API_VERSION: u32 = 1;
+
+#[derive(Debug, Clone)]
+pub struct PluginMetadata {
+    pub name: String,
+    pub version: String,
+    pub description: String,
+}
+
+pub trait PluginInstance {
+    fn init(&mut self, config: &Value) -> Result<(), String>;
+    fn execute(&self, input: &str) -> String;
+    fn metadata(&self) -> PluginMetadata;
+}
+
+#[repr(C)]
+pub struct PluginInterface {
+    pub create_plugin: unsafe extern "C" fn() -> *mut dyn PluginInstance,
+    pub destroy_plugin: unsafe extern "C" fn(*mut dyn PluginInstance),
+    pub api_version: u32,
+}
+
+pub struct PluginManager {
+    plugins: HashMap<String, (Box<dyn PluginInstance>, Library)>,
+}
+
+#[derive(Debug)]
+pub enum PluginError {
+    LibraryLoad(libloading::Error),
+    SymbolError(libloading::Error),
+    IncompatibleAPI,
+    NotFound,
+}
+
+impl fmt::Display for PluginError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            PluginError::LibraryLoad(e) => write!(f, "Library load error: {}", e),
+            PluginError::SymbolError(e) => write!(f, "Symbol error: {}", e),
+            PluginError::IncompatibleAPI => write!(f, "Incompatible API version"),
+            PluginError::NotFound => write!(f, "Plugin not found"),
+        }
+    }
+}
+
+impl PluginManager {
+    pub fn new() -> Self {
+        Self {
+            plugins: HashMap::new(),
+        }
+    }
+
+    pub fn load_plugin(&mut self, path: &Path) -> Result<(), PluginError> {
+        unsafe {
+            log::debug!("Loading plugin from: {}", path.display());
+
+            let lib = Library::new(path).map_err(PluginError::LibraryLoad)?;
+
+            println!("------------1");
+
+            let interface_ptr: Symbol<unsafe extern "C" fn() -> &'static PluginInterface> =
+                lib.get(b"get_plugin_interface").map_err(PluginError::SymbolError)?;
+            let interface = interface_ptr();
+
+            assert_eq!(
+                std::mem::size_of::<PluginInterface>(),
+                std::mem::size_of_val(&*interface),
+                "ABI size mismatch"
+            );
+
+            println!("------------2");
+
+            if interface.api_version != CURRENT_API_VERSION {
+                return Err(PluginError::IncompatibleAPI);
+            }
+
+            let plugin_ptr = (interface.create_plugin)();
+            let mut plugin = Box::from_raw(plugin_ptr);
+
+            plugin.init(&Value::Null).map_err(|e| {
+                PluginError::SymbolError(libloading::Error::DlSymUnknown)
+            })?;
+
+            let metadata = plugin.metadata();
+            self.plugins.insert(metadata.name.clone(), (plugin, lib));
+
+            Ok(())
+        }
+    }
+
+    pub fn unload_plugin(&mut self, name: &str) -> Result<(), PluginError> {
+        let (plugin, lib) = self.plugins.remove(name).ok_or(PluginError::NotFound)?;
+
+        unsafe {
+            let interface: Symbol<PluginInterface> = lib
+                .get(b"plugin_interface")
+                .map_err(PluginError::SymbolError)?;
+
+            (interface.destroy_plugin)(Box::into_raw(plugin));
+        }
+
+
+        Ok(())
+    }
+
+    pub fn execute_plugin(&self, name: &str, input: &str) -> Result<String, String> {
+        self.plugins
+            .get(name)
+            .map(|(p, _)| p.execute(input))
+            .ok_or_else(|| format!("Plugin {} not found", name))
+    }
+
+    pub fn get_metadata(&self, name: &str) -> Option<PluginMetadata> {
+        self.plugins.get(name).map(|(p, _)| p.metadata())
+    }
+}
+
+impl Drop for PluginManager {
+    fn drop(&mut self) {
+        let plugins = std::mem::take(&mut self.plugins);
+        for (name, (plugin, lib)) in plugins.into_iter() {
+            unsafe {
+                if let Ok(interface) = lib.get::<PluginInterface>(b"plugin_interface") {
+                    log::debug!("Dropping plugin: {}", name);
+                    (interface.destroy_plugin)(Box::into_raw(plugin));
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::env;
+    use std::path::PathBuf;
+    use std::sync::{Once, OnceLock};
+
+    static INIT: Once = Once::new();
+    static PLUGIN_PATH: OnceLock<PathBuf> = OnceLock::new();
+
+    fn init_logger() {
+        INIT.call_once(|| {
+            env_logger::builder()
+                .filter_level(log::LevelFilter::Debug)
+                .init();
+        });
+    }
+
+    fn get_plugin_path() -> &'static Path {
+        PLUGIN_PATH.get_or_init(|| {
+            let target_dir = PathBuf::from("/Users/bytedance/Documents/opensource/lance/target");
+
+            let mut path = target_dir.join("debug").join("examples");
+
+            #[cfg(target_os = "linux")]
+            path.push("libtest_plugin.so");
+            #[cfg(target_os = "macos")]
+            path.push("libtest_plugin.dylib");
+            #[cfg(target_os = "windows")]
+            path.push("test_plugin.dll");
+
+            assert!(path.exists(), "Plugin not found at: {}", path.display());
+            path
+        })
+    }
+
+    #[test]
+    fn test_load_valid_plugin() {
+        init_logger();
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+
+        let result = manager.load_plugin(path);
+        assert!(result.is_ok(), "Load failed: {:?}", result.err());
+
+        let metadata = manager.get_metadata("test_plugin").unwrap();
+        assert_eq!(metadata.version, "1.0");
+    }
+
+    #[test]
+    fn test_load_nonexistent_library() {
+        let mut manager = PluginManager::new();
+        let path = Path::new("non_existent_plugin.so");
+
+        let result = manager.load_plugin(path);
+        assert!(
+            matches!(result, Err(PluginError::LibraryLoad(_))),
+            "Expected library load error"
+        );
+    }
+
+    #[test]
+    fn test_execute_plugin() {
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+        manager.load_plugin(path).unwrap();
+
+        let result = manager.execute_plugin("test_plugin", "test_input");
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), "Processed: test_input");
+    }
+
+    #[test]
+    fn test_execute_nonexistent_plugin() {
+        let manager = PluginManager::new();
+
+        let result = manager.execute_plugin("nonexistent_plugin", "input");
+        assert!(
+            result.is_err(),
+            "Should return error for nonexistent plugin"
+        );
+    }
+
+    #[test]
+    fn test_unload_plugin() {
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+        manager.load_plugin(path).unwrap();
+
+        let result = manager.unload_plugin("test_plugin");
+        assert!(result.is_ok(), "Unload failed");
+        assert!(
+            manager.get_metadata("test_plugin").is_none(),
+            "Plugin metadata still present after unload"
+        );
+    }
+
+    #[test]
+    fn test_drop_cleanup() {
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+        manager.load_plugin(path).unwrap();
+
+        drop(manager);
+    }
+
+    #[test]
+    fn test_metadata_retrieval() {
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+        manager.load_plugin(path).unwrap();
+
+        let metadata = manager.get_metadata("test_plugin").unwrap();
+        assert_eq!(metadata.description, "Test Plugin");
+    }
+
+    #[test]
+    fn test_reload_same_plugin() {
+        let mut manager = PluginManager::new();
+        let path = get_plugin_path();
+
+        manager.load_plugin(path).unwrap();
+        let first_load_count = manager.plugins.len();
+
+        manager.load_plugin(path).unwrap();
+        assert_eq!(
+            manager.plugins.len(),
+            first_load_count,
+            "Reloading same plugin should replace existing entry"
+        );
+    }
+}

rust/lance/src/plugin/test_plugin.rs

@@ -0,0 +1,45 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use serde_json::Value;
+use lance::plugin::{PluginInstance, PluginInterface, PluginMetadata};
+
+pub struct TestPlugin;
+
+impl PluginInstance for TestPlugin {
+    fn init(&mut self, _: &Value) -> Result<(), String> {
+        Ok(())
+    }
+
+    fn execute(&self, input: &str) -> String {
+        format!("Processed: {}", input)
+    }
+
+    fn metadata(&self) -> PluginMetadata {
+        PluginMetadata {
+            name: "test_plugin".into(),
+            version: "1.0".into(),
+            description: "Test Plugin".into(),
+        }
+    }
+}
+
+#[no_mangle]
+pub extern "C" fn create() -> *mut dyn PluginInstance {
+    Box::into_raw(Box::new(TestPlugin))
+}
+
+#[no_mangle]
+pub extern "C" fn destroy(plugin: *mut dyn PluginInstance) {
+    unsafe { Box::from_raw(plugin) };
+}
+
+
+#[no_mangle]
+pub extern "C" fn get_plugin_interface() -> &'static PluginInterface {
+    &PluginInterface {
+        create_plugin: create,
+        destroy_plugin: destroy,
+        api_version: 1,
+    }
+}