feat(metactl): add lua grpc client functionality (#18438)

Extend lua subcommand with gRPC client capabilities for meta operations:
- Add new_grpc_client() function to create gRPC client instances
- Implement async get() and upsert() methods with tuple error handling
- Export NULL constant to Lua environment for proper null handling
- Add comprehensive test suite for gRPC client functionality
- Support both stdin and file input modes for Lua scripts

The gRPC client returns (result, error) tuples following Lua conventions
and converts Rust structures to native Lua tables for better usability.

Example usage:
```lua
local client = new_grpc_client("127.0.0.1:9191")

-- Store a key-value pair
local result, err = client:upsert("my_key", "my_value")
if err then
    print("Error:", err)
else
    print("Stored:", to_string(result))
end

-- Retrieve the value
local value, err = client:get("my_key")
if err then
    print("Error:", err)
else
    print("Retrieved:", to_string(value))
end
```

Run with: `databend-metactl lua --file script.lua`
Or pipe: `echo 'script content' | databend-metactl lua`

Author: drmingdrmer

Cargo.lock

@@ -387,7 +387,7 @@ match-template = "0.0.1"
 md-5 = "0.10.5"
 memchr = { version = "2", default-features = false }
 micromarshal = "0.7.0"
-mlua = { version = "0.11", features = ["lua54", "vendored"] }
+mlua = { version = "0.11", features = ["lua54", "vendored", "async", "serialize"] }
 mockall = "0.11.2"
 mysql_async = { version = "0.34", default-features = false, features = ["native-tls-tls"] }
 naive-cityhash = "0.2.0"

Cargo.toml

@@ -54,6 +54,10 @@ use databend_meta::version::METASRV_COMMIT_VERSION;
 use display_more::DisplayOptionExt;
 use futures::stream::TryStreamExt;
 use mlua::Lua;
+use mlua::LuaSerdeExt;
+use mlua::UserData;
+use mlua::UserDataMethods;
+use mlua::Value;
 use serde::Deserialize;
 
 #[derive(Debug, Deserialize, Parser)]
@@ -233,6 +237,34 @@ impl App {
     async fn run_lua(&self, args: &LuaArgs) -> anyhow::Result<()> {
         let lua = Lua::new();
 
+        // Register new_grpc_client function
+        let new_grpc_client = lua
+            .create_function(|_lua, address: String| {
+                let client = MetaGrpcClient::try_create(
+                    vec![address],
+                    "root",
+                    "xxx",
+                    Some(Duration::from_secs(2)),
+                    Some(Duration::from_secs(1)),
+                    None,
+                )
+                .map_err(|e| {
+                    mlua::Error::external(format!("Failed to create gRPC client: {}", e))
+                })?;
+
+                Ok(LuaGrpcClient::new(client))
+            })
+            .map_err(|e| anyhow::anyhow!("Failed to create new_grpc_client function: {}", e))?;
+
+        lua.globals()
+            .set("new_grpc_client", new_grpc_client)
+            .map_err(|e| anyhow::anyhow!("Failed to register new_grpc_client: {}", e))?;
+
+        // Export NULL constant to Lua environment
+        lua.globals()
+            .set("NULL", Value::NULL)
+            .map_err(|e| anyhow::anyhow!("Failed to register NULL constant: {}", e))?;
+
         let script = match &args.file {
             Some(path) => std::fs::read_to_string(path)?,
             None => {
@@ -242,7 +274,7 @@ impl App {
             }
         };
 
-        if let Err(e) = lua.load(&script).exec() {
+        if let Err(e) = lua.load(&script).exec_async().await {
             return Err(anyhow::anyhow!("Lua execution error: {}", e));
         }
         Ok(())
@@ -264,6 +296,44 @@ impl App {
     }
 }
 
+struct LuaGrpcClient {
+    client: Arc<ClientHandle>,
+}
+
+impl LuaGrpcClient {
+    fn new(client: Arc<ClientHandle>) -> Self {
+        Self { client }
+    }
+}
+
+impl UserData for LuaGrpcClient {
+    fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
+        methods.add_async_method("get", |lua, this, key: String| async move {
+            match this.client.get_kv(&key).await {
+                Ok(result) => match lua.to_value(&result) {
+                    Ok(lua_value) => Ok((Some(lua_value), None::<String>)),
+                    Err(e) => Ok((None::<Value>, Some(format!("Lua conversion error: {}", e)))),
+                },
+                Err(e) => Ok((None::<Value>, Some(format!("gRPC error: {}", e)))),
+            }
+        });
+
+        methods.add_async_method(
+            "upsert",
+            |lua, this, (key, value): (String, String)| async move {
+                let upsert = UpsertKV::update(key, value.as_bytes());
+                match this.client.request(upsert).await {
+                    Ok(result) => match lua.to_value(&result) {
+                        Ok(lua_value) => Ok((Some(lua_value), None::<String>)),
+                        Err(e) => Ok((None::<Value>, Some(format!("Lua conversion error: {}", e)))),
+                    },
+                    Err(e) => Ok((None::<Value>, Some(format!("gRPC error: {}", e)))),
+                }
+            },
+        );
+    }
+}
+
 #[derive(Debug, Clone, Deserialize, Subcommand)]
 enum CtlCommand {
     Status(StatusArgs),

src/meta/binaries/metactl/main.rs

@@ -0,0 +1,145 @@
+-- Function to check if a value is NULL (mlua NULL constant)
+function is_null(value)
+    return value == NULL
+end
+
+-- Function to normalize NULL values to nil for easier handling
+function normalize_value(value)
+    if is_null(value) then
+        return nil
+    end
+    return value
+end
+
+-- Function to check if a table represents a bytes vector and convert it to string
+function bytes_vector_to_string(value)
+    -- Check if table represents a bytes vector (integer indices starting from 1, u8 values)
+    local max_index = 0
+    local min_index = math.huge
+    
+    -- First pass: check if all keys are integers and find range
+    for k, v in pairs(value) do
+        if type(k) ~= "number" or k ~= math.floor(k) or k < 1 then
+            return nil -- Not a bytes vector
+        end
+        if type(v) ~= "number" or v ~= math.floor(v) or v < 0 or v > 255 then
+            return nil -- Not u8 values
+        end
+        max_index = math.max(max_index, k)
+        min_index = math.min(min_index, k)
+    end
+    
+    -- Check if indices are consecutive starting from 1
+    if min_index == 1 then
+        local expected_length = max_index
+        local actual_length = 0
+        for _ in pairs(value) do
+            actual_length = actual_length + 1
+        end
+        if actual_length == expected_length then
+            -- Check size limit to prevent memory exhaustion (max 1MB)
+            if max_index > 1048576 then
+                return '"<bytes vector too large: ' .. max_index .. ' bytes>"'
+            end
+            
+            -- Convert bytes vector to string
+            local chars = {}
+            for i = 1, max_index do
+                chars[i] = string.char(value[i])
+            end
+            return '"' .. table.concat(chars) .. '"'
+        end
+    end
+    
+    return nil -- Not a valid bytes vector
+end
+
+-- Function to convert a value or table into a single-line string recursively
+function to_string(value)
+    if value == nil then
+        return "nil"
+    end
+    
+    if is_null(value) then
+        return "NULL"
+    end
+
+    if type(value) == "boolean" then
+        return tostring(value)
+    end
+
+    if type(value) == "number" then
+        return tostring(value)
+    end
+
+    if type(value) == "string" then
+        -- Escape quotes in the string
+        return '"' .. string.gsub(value, '"', '\\"') .. '"'
+    end
+
+    if type(value) == "table" then
+        -- Check if it's a bytes vector first
+        local bytes_string = bytes_vector_to_string(value)
+        if bytes_string then
+            return bytes_string
+        end
+        
+        -- Regular table processing
+        local result = "{"
+        local keys = {}
+
+        -- Collect all keys
+        for k in pairs(value) do
+            table.insert(keys, k)
+        end
+
+        -- Sort keys if they're all strings or numbers
+        local can_sort = true
+        for _, k in ipairs(keys) do
+            if type(k) ~= "string" and type(k) ~= "number" then
+                can_sort = false
+                break
+            end
+        end
+
+        if can_sort then
+            table.sort(keys, function(a, b)
+                if type(a) == type(b) then
+                    return a < b
+                else
+                    -- Put numbers before strings
+                    return type(a) == "number"
+                end
+            end)
+        end
+
+        -- Process each key-value pair
+        local first = true
+        for _, k in ipairs(keys) do
+            local v = value[k]
+            
+            -- Skip nil and NULL values
+            if v ~= nil and not is_null(v) then
+                if not first then
+                    result = result .. ","
+                end
+                first = false
+
+                local key_str
+                if type(k) == "string" then
+                    key_str = '"' .. string.gsub(k, '"', '\\"') .. '"'
+                else
+                    key_str = "[" .. tostring(k) .. "]"
+                end
+
+                result = result .. key_str .. "=" .. to_string(v)
+            end
+        end
+
+        result = result .. "}"
+        return result
+    end
+
+    -- Handle function, userdata, thread
+    return "<" .. type(value) .. ">"
+end