Rust 中的预览功能 - VillageSQL Server for MySQL

预览功能 API 并不稳定。基于预览功能构建的扩展程序在服务器更新后可能无法加载。

预览功能是服务器提供的功能，在最终确定其 API 之前，会将其暴露给扩展。Rust SDK 封装了其中五项：状态变量、系统变量、后台线程工作器、从工作线程发起的 SQL 查询和密钥环访问。本页介绍如何从 Rust 声明和使用其中的每一项。有关该概念、预览层以及仅在 C++ 中存在的功能，请参阅预览功能。

前提条件

使用任何预览功能的扩展，只有在 vsql_allow_preview_extensions 为 ON 时才能安装：

SET PERSIST vsql_allow_preview_extensions = ON;

有关详细信息，包括为什么此变量会拒绝 SET GLOBAL，请参阅启用预览层。您还需要一套可用的 Rust 扩展设置——使用 Rust 构建扩展演练涵盖了工具链、cargo-vsql 以及您的第一个 extension! 块。

Rust SDK 封装了什么

功能	Rust 模块	SDK 示例
`vsql::status_var`	`villagesql::preview::status_var`	`vsql_status_var`
`vsql::sys_var`	`villagesql::preview::sys_var`	`vsql_sys_var`
`vsql::preview::thread_worker`	`villagesql::preview::thread_worker`	`vsql_thread_worker`
`vsql::preview::sql_query`	`villagesql::preview::sql_query`	`vsql_sql_query`
`vsql::preview::keyring`	`villagesql::preview::keyring`	`vsql_keyring`

其余的预览功能——auth、mysql_services、session_var、statement_event、自定义索引功能以及列存储 ABI——目前仅支持 C++。如果您需要其中之一，请使用 C++ SDK。

注册模式

将每个功能声明为 static，然后在 extension! 的 requires: 部分中按引用列出它——这相当于 C++ SDK 中的 .with()：

use villagesql::preview::keyring::KeyringCapability;

static KEYRING: KeyringCapability = KeyringCapability::new();

villagesql::extension! {
    funcs: [
        // ...
    ],
    requires: [
        &KEYRING,
    ]
}

服务器在加载时填充功能对象；在此之前，其访问器方法会报告该功能不可用，而不是崩溃。static 是必需的——服务器在扩展的整个生命周期内都持有指向该功能的指针。

状态变量

status_var 功能通过 SHOW GLOBAL STATUS 公开由扩展拥有的计数器。您的扩展以 'static 原子类型拥有存储空间并写入其中；服务器每次查询状态变量时，都会通过指针读取。将每个变量声明为一个 StatusVarSpec——Int 由 AtomicI64 支持，或 Double 由 SDK 的 AtomicF64 支持（Rust 标准库没有原子 f64，因此 SDK 提供了一个，带有 new、load 和 store）：

use std::sync::atomic::{AtomicI64, Ordering};

use villagesql::preview::status_var::{AtomicF64, StatusVarCapability, StatusVarSpec};
use villagesql::{InValue, VdfReturn};

static REQUESTS: AtomicI64 = AtomicI64::new(0);
static LOAD: AtomicF64 = AtomicF64::new(0.5);

static SPECS: &[StatusVarSpec] = &[
    StatusVarSpec::Int {
        name: c"requests",
        value: &REQUESTS,
    },
    StatusVarSpec::Double {
        name: c"load",
        value: &LOAD,
    },
];

static STATUS_VAR: StatusVarCapability = StatusVarCapability::new(SPECS);

fn bump_impl(_args: &[InValue]) -> VdfReturn {
    let n = REQUESTS.fetch_add(1, Ordering::Relaxed) + 1;
    VdfReturn::int(n)
}

villagesql::extension! {
    funcs: [
        villagesql::func!(bump_impl, "bump", [] -> villagesql::Type::Int),
    ],
    requires: [
        &STATUS_VAR,
    ]
}

执行 INSTALL EXTENSION 之后，这些变量会以扩展名称作为前缀出现：

SELECT vsql_status_var.bump();
-- 1
SHOW GLOBAL STATUS LIKE 'vsql_status_var%';

Variable_name	Value
vsql_status_var.load	0.500000
vsql_status_var.requests	1

系统变量

sys_var 功能注册由您的扩展拥有的 MySQL 系统变量。支持三种类型：Bool、Int（带 min/max 边界）和 Str。每个描述符都携带一个名称、一条注释、一个默认值以及一个可选的 on_change 回调。名称和字符串默认值是 &'static CStr 值——请将它们写为 C 字符串字面量（c"enabled"）：

use villagesql::preview::sys_var::{SysVarCapability, SysVarSpec};

static SPECS: &[SysVarSpec] = &[
    SysVarSpec::Bool {
        name: c"enabled",
        comment: c"Enable the feature",
        default: true,
        on_change: None,
    },
    SysVarSpec::Int {
        name: c"threshold",
        comment: c"Threshold in milliseconds",
        default: 1000,
        min: 0,
        max: 60000,
        on_change: None,
    },
    SysVarSpec::Str {
        name: c"log_path",
        comment: c"Path to the log file",
        default: c"/tmp/vsql_sys_var.log",
        on_change: None,
    },
];

static SYS_VAR: SysVarCapability = SysVarCapability::new(SPECS);

villagesql::extension! {
    funcs: [],
    requires: [
        &SYS_VAR,
    ]
}

即使 funcs: 部分为空，它也必须出现在 requires: 之前。安装之后，可以使用扩展名称作为前缀来访问这些变量：

SELECT @@global.vsql_sys_var.enabled;
-- 1
SELECT @@global.vsql_sys_var.threshold;
-- 1000
SET GLOBAL vsql_sys_var.enabled = 0;

响应变更

on_change 是一个原始 C 回调，由服务器在变量被设置之后调用。服务器在持有其全局系统变量锁时调用它，因此请让它保持快速，并且不要 panic——此处的 panic 会跨越 FFI 边界。该回调接收来自原始 ABI 层的 *const vef_sys_var_change_t：

use std::sync::atomic::{AtomicU64, Ordering};
use villagesql::sys::vef_sys_var_change_t;

static CHANGE_COUNT: AtomicU64 = AtomicU64::new(0);

unsafe extern "C" fn on_enabled_change(_change: *const vef_sys_var_change_t) {
    CHANGE_COUNT.fetch_add(1, Ordering::Relaxed);
}

调用 SysVarCapability::get 或 SysVarCapability::set、运行 SQL，或等待一个执行上述任一操作的线程，都会在该锁上造成死锁。请让回调只在您自己的静态变量中做记账工作（如上所示），并将需要 SQL 的工作交给线程工作器。

使用 on_change: Some(on_enabled_change) 将它接入某个描述符。

从扩展代码读取和写入

SysVarCapability 还公开了 get() 和 set()，用于通过服务器进行编程访问（因此范围验证和持久化都会为您处理）。set() 接受一个 scope 参数来选择持久化方式：null 只更改运行中的值，因此它在重启后会回退；"PERSIST" 更改运行中的值并将其写入持久化配置；"PERSIST_ONLY" 写入持久化配置而不改动运行中的值，因此它在下次重启时生效。两者都是接受 NUL 结尾 C 字符串的 unsafe FFI 方法，并且两者都使用相反的 C 约定：Some(false) 表示成功，Some(true) 表示服务器报告了错误，而 None 表示该功能不可用。在 get 成功时，服务器会写入一个 malloc 分配的字符串，您必须使用 C 的 free() 释放它。vsql_sys_var 示例展示了完整的模式，包括 free extern 和安全性注释。

线程工作器

thread_worker 功能在服务器管理的后台线程上运行您提供的函数。服务器在加载时注册一个控制系统变量；当它为 ON 时，您的工作函数会在定期计时器上、在文件描述符准备就绪时，或在启用/禁用转换时被调用。您的工作函数是普通的安全 Rust 代码：

use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;

use villagesql::preview::thread_worker::{
    NextWakeup, ThreadHandle, ThreadWorkerCapability, WakeupReason,
};
use villagesql::{InValue, VdfReturn};

static TICKS: AtomicI64 = AtomicI64::new(0);

fn worker(reason: WakeupReason, _handle: &ThreadHandle) -> NextWakeup {
    if reason == WakeupReason::Periodic {
        TICKS.fetch_add(1, Ordering::Relaxed);
    }
    NextWakeup::unchanged()
}

static WORKER: ThreadWorkerCapability =
    ThreadWorkerCapability::new(worker, "ticker", Duration::from_millis(100), None);

fn ticks_impl(_args: &[InValue]) -> VdfReturn {
    VdfReturn::int(TICKS.load(Ordering::Relaxed))
}

villagesql::extension! {
    funcs: [
        villagesql::func!(ticks_impl, "ticks", [] -> villagesql::Type::Int),
    ],
    requires: [
        &WORKER,
    ]
}

ThreadWorkerCapability::new 接受工作函数、一个线程名称后缀、初始睡眠间隔，以及一个可选的控制变量名称覆盖值。当该覆盖值为 None 时，控制变量被命名为 {suffix}_enabled，并注册在扩展的前缀之下：

SHOW GLOBAL VARIABLES LIKE '%ticker%';

Variable_name	Value
vsql_thread_worker.ticker_enabled	OFF

SET GLOBAL vsql_thread_worker.ticker_enabled = ON;
SELECT SLEEP(0.5);
SELECT vsql_thread_worker.ticks();
-- 4  (varies with timing — one tick per 100 ms while enabled)
SET GLOBAL vsql_thread_worker.ticker_enabled = OFF;

唤醒

WakeupReason 告诉您服务器为什么进行调用：Enable、Periodic、PollFd 或 Disable。您的返回值会调整下一次唤醒：

NextWakeup::unchanged()——保持当前的睡眠间隔和轮询文件描述符。
NextWakeup::after(duration)——在 duration 之后再次唤醒。
将 poll_fd 字段设置为一个大于零的文件描述符，以便在它变为可读时也唤醒；或设置为 -1 以清除先前设置的文件描述符。

长度为零的 Duration 会退化为“无变化”——底层 C ABI 将 0 保留给该含义，因此无法表达立即唤醒。如果工作函数发生 panic，SDK 会在 FFI 边界捕获该 panic，并将该次调用视为返回了 NextWakeup::unchanged()——工作器会继续运行。将 handle 参数传给 SqlQueryCapability::open，即可在工作器内部执行 SQL。

SQL 查询

sql_query 功能从 thread_worker 回调中打开一个 SQL 会话，因此后台工作器可以对加载它的服务器运行语句。将它与 ThreadWorkerCapability 一起声明，并使用工作器收到的 ThreadHandle 打开会话：

use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;

use villagesql::preview::sql_query::SqlQueryCapability;
use villagesql::preview::thread_worker::{
    NextWakeup, ThreadHandle, ThreadWorkerCapability, WakeupReason,
};
use villagesql::{InValue, VdfReturn};

static USER_COUNT: AtomicI64 = AtomicI64::new(-1);

static SQL_QUERY: SqlQueryCapability = SqlQueryCapability::new();

fn worker(reason: WakeupReason, handle: &ThreadHandle) -> NextWakeup {
    if reason == WakeupReason::Periodic {
        if let Some(session) = SQL_QUERY.open(handle) {
            if let Some(mut result) = session.execute("SELECT COUNT(*) FROM mysql.user") {
                if !result.has_error() {
                    if let Some(n) = result.next_row().and_then(|row| row.get_int(0)) {
                        USER_COUNT.store(n, Ordering::Relaxed);
                    }
                }
            }
        }
    }
    NextWakeup::unchanged()
}

static WORKER: ThreadWorkerCapability =
    ThreadWorkerCapability::new(worker, "user_count", Duration::from_secs(5), None);

fn user_count_impl(_args: &[InValue]) -> VdfReturn {
    VdfReturn::int(USER_COUNT.load(Ordering::Relaxed))
}

villagesql::extension! {
    funcs: [
        villagesql::func!(user_count_impl, "user_count", [] -> villagesql::Type::Int),
    ],
    requires: [
        &WORKER,
        &SQL_QUERY,
    ]
}

如果该功能未被填充（预览功能关闭或未请求）、ABI 过旧，或服务器未能打开会话，SqlQueryCapability::open(handle) 会返回 None——请在使用结果之前检查它。 Session 提供两种运行语句的方式：

execute(sql) 缓冲完整的结果集并返回一个 QueryResult。反复调用 next_row() 直到它返回 None；每个 Row 通过 get_int(i)、get_str(i)、get_real(i) 或 get_bytes(i) 读取列。get_bytes/get_str 在 SQL NULL 或索引超出范围时返回 None（get_str 遇到非 UTF-8 字节时也是如此）；get_int/get_real 会先将列解析为文本，因此在文本无法解析时也会返回 None。execute 本身只在设置失败（没有会话或内存不足）时返回 None。
for_each(sql, |row: Row<'_>| { ... }) 将各行流式传给一个闭包，而不是缓冲完整的结果集。它返回的 QueryResult 只携带诊断信息——没有可迭代的缓冲行——并且同样在设置失败时返回 None。

在信任语句已成功之前，请检查任一结果上的 has_error() 和 error()（一个带有 errno、severity、sqlstate 和 message 的 Option<Diag>）。warnings() 返回附加在结果上的所有警告和提示。 Session 和 QueryResult 各自在被丢弃时关闭其底层句柄——没有需要您自行调用的关闭方法。

密钥环访问

keyring 功能读取和写入存储在 MySQL 密钥环组件中的密钥——API 密钥、加密密钥，以及任何不应存储在表中的内容。密钥环组件（例如 component_keyring_file）必须按照 Keyring Component Installation 所述，在服务器启动时通过组件清单及其配置文件加载；INSTALL COMPONENT 不会使它可用。如果没有密钥环组件，每次读取和写入都会以 KeyringError::NoComponent 失败。

use std::ffi::CString;

use villagesql::preview::keyring::KeyringCapability;
use villagesql::{InValue, VdfReturn};

static KEYRING: KeyringCapability = KeyringCapability::new();

const MAX_SECRET_LEN: usize = 1024;

fn keyring_read_impl(args: &[InValue]) -> VdfReturn {
    let Some(&InValue::String(data_id)) = args.first() else {
        return VdfReturn::null();
    };
    let Ok(data_id) = CString::new(data_id) else {
        return VdfReturn::null();
    };

    let mut buf = [0u8; MAX_SECRET_LEN];
    match KEYRING.read(&data_id, None, &mut buf) {
        Ok(Some(n)) => match std::str::from_utf8(&buf[..n]) {
            Ok(s) => VdfReturn::string(s),
            Err(_) => VdfReturn::null(),
        },
        Ok(None) => VdfReturn::null(),
        Err(e) => VdfReturn::error(format!("keyring: {e}")),
    }
}

villagesql::extension! {
    funcs: [
        villagesql::func!(
            keyring_read_impl, "keyring_read",
            [villagesql::Type::String] -> villagesql::Type::String,
            buffer_size: MAX_SECRET_LEN
        ),
    ],
    requires: [
        &KEYRING,
    ]
}

read(data_id, auth_id, buf) 会填充您传入的缓冲区，并返回带有密钥长度的 Ok(Some(n))；或者当 data_id 下不存在密钥时返回 Ok(None)——这是一种正常结果，而不是错误。write(data_id, auth_id, data) 在成功时返回 Ok(())。auth_id 是所有者用户；对于与特定用户无关的内部密钥，请传递 None。两者在失败时都返回 Err(KeyringError)：CapabilityUnavailable（该功能从未被接入）、NoComponent（服务器上没有密钥环组件）或 Other。密钥环没有大小探测：如果某个密钥大于您传给 read 的缓冲区，返回结果为 Ok(None)，与密钥缺失无法区分。请按您预期存储的最大密钥来确定缓冲区大小。

后续步骤

预览功能 (C++)

完整的功能索引、预览层，以及仅支持 C++ 的功能：auth、mysql_services、session_var、statement_event、自定义索引功能和列存储。

Rust API 参考

InValue、VdfReturn、extension!、func! 和 custom_type!——所有字段。

​前提条件

​Rust SDK 封装了什么

​注册模式

​状态变量

​系统变量

​响应变更

​从扩展代码读取和写入

​线程工作器

​唤醒

​SQL 查询

​密钥环访问

​后续步骤

预览功能 (C++)

Rust API 参考

前提条件

Rust SDK 封装了什么

注册模式

状态变量

系统变量

响应变更

从扩展代码读取和写入

线程工作器

唤醒

SQL 查询

密钥环访问

后续步骤