From ec20a0de73df8f16a1358f32cb092f0a1190f283 Mon Sep 17 00:00:00 2001
From: Igor Kuznetsov <kuznetsovin@gmail.com>
Date: Wed, 17 Nov 2021 14:06:02 +0300
Subject: [PATCH] feat: add query execute function
Add sql executor function binding.
---
src/cluster_lua.rs | 31 +++++++++-------
src/parser.rs | 49 +++++++++++++++++++++++---
test_app/app/roles/api.lua | 13 ++++++-
test_app/app/roles/storage.lua | 5 +++
test_app/test/integration/api_test.lua | 21 +++++++++--
5 files changed, 99 insertions(+), 20 deletions(-)
diff --git a/src/cluster_lua.rs b/src/cluster_lua.rs
index 4ac46370bf..015333fd5a 100644
--- a/src/cluster_lua.rs
+++ b/src/cluster_lua.rs
@@ -2,23 +2,27 @@ use std::ffi::CStr;
use std::os::raw::c_char;
use tarantool::ffi::lua::{
- lua_State, lua_getglobal, lua_pushinteger, lua_setfield, lua_settop, lua_tointeger,
- lua_tostring, LUA_GLOBALSINDEX,
+ lua_State, lua_getglobal, lua_pushinteger, lua_pushstring, lua_setfield, lua_settop,
+ lua_tointeger, lua_tostring, LUA_GLOBALSINDEX,
};
const LUA_FUNCS: &str = "local cartridge = require('cartridge');
local vshard = require('vshard')
+local yaml = require('yaml')
-function get_server_uri_by_bucket(bucket_id)
- local replicaset = vshard.router.route(bucket_id)
+function execute_sql(bucket_id, query)
+ local data, err = vshard.router.call(
+ bucket_id,
+ 'read',
+ 'box.execute',
+ { query }
+ )
- if replicaset['master'] ~= nil then
- if replicaset.master['conn'] ~= nil then
- return replicaset.master.conn.host .. ':' .. replicaset.master.conn.port
- end
- end
+ if err ~= nil then
+ error(err)
+ end
- return ''
+ return yaml.encode(data)
end
function get_cluster_schema()
@@ -55,13 +59,14 @@ pub fn init_cluster_functions(l: *mut lua_State) {
}
#[allow(dead_code)]
-pub fn get_server_uri_by_bucket_id(l: *mut lua_State, bucket_id: isize) -> String {
+pub fn execute_sql(l: *mut lua_State, bucket_id: isize, query: &str) -> String {
unsafe {
- lua_getglobal(l, crate::c_ptr!("get_server_uri_by_bucket"));
+ lua_getglobal(l, crate::c_ptr!("execute_sql"));
lua_pushinteger(l, bucket_id);
+ lua_pushstring(l, query.as_ptr().cast::<i8>());
}
- let res = unsafe { lua_pcall(l, 1, 1, 0) };
+ let res = unsafe { lua_pcall(l, 2, 1, 0) };
if res != 0 {
panic!("{} {:?}", res, unsafe {
CStr::from_ptr(lua_tostring(l, -1))
diff --git a/src/parser.rs b/src/parser.rs
index 6c9b0ab0fb..ab0d6ef227 100644
--- a/src/parser.rs
+++ b/src/parser.rs
@@ -1,5 +1,5 @@
use crate::bucket::str_to_bucket_id;
-use crate::cluster_lua::{get_cluster_schema, init_cluster_functions};
+use crate::cluster_lua::{execute_sql, get_cluster_schema, init_cluster_functions};
use crate::errors::QueryPlannerError;
use crate::query::ParsedTree;
use crate::schema::Cluster;
@@ -9,10 +9,11 @@ use std::cell::RefCell;
use std::fmt;
use std::os::raw::c_int;
use tarantool::error::TarantoolErrorCode;
-use tarantool::ffi::lua::luaT_state;
+use tarantool::ffi::lua::{luaT_state, lua_State};
use tarantool::tuple::{AsTuple, FunctionArgs, FunctionCtx, Tuple};
thread_local!(static CARTRIDGE_SCHEMA: RefCell<Cluster> = RefCell::new(Cluster::new()));
+thread_local!(static LUA_STATE: RefCell<*mut lua_State> = RefCell::new( unsafe { luaT_state() }));
#[derive(Serialize, Deserialize)]
struct Args {
@@ -30,8 +31,7 @@ pub extern "C" fn parse_sql(ctx: FunctionCtx, args: FunctionArgs) -> c_int {
let args: Tuple = args.into();
let args = args.into_struct::<Args>().unwrap();
- let l = unsafe { luaT_state() };
- init_cluster_functions(l);
+ let l = LUA_STATE.try_with(|s| s.clone().into_inner()).unwrap();
CARTRIDGE_SCHEMA.with(|s| {
let mut schema = s.clone().into_inner();
@@ -88,6 +88,47 @@ pub extern "C" fn calculate_bucket_id(ctx: FunctionCtx, args: FunctionArgs) -> c
0
}
+#[derive(Debug, Serialize, Deserialize)]
+struct ExecQueryArgs {
+ pub bucket_id: isize,
+ pub query: String,
+}
+
+impl AsTuple for ExecQueryArgs {}
+
+#[no_mangle]
+pub extern "C" fn execute_query(ctx: FunctionCtx, args: FunctionArgs) -> c_int {
+ let args: Tuple = args.into();
+ let args = args.into_struct::<ExecQueryArgs>().unwrap();
+
+ let l = LUA_STATE.try_with(|s| s.clone().into_inner()).unwrap();
+ let result = execute_sql(l, args.bucket_id, &args.query);
+ ctx.return_mp(&result).unwrap();
+ 0
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct InitArgs {
+ pub login: String,
+ pub password: String,
+}
+
+impl AsTuple for InitArgs {}
+
+/**
+Function must be called in router and storage roles of cartridge application.
+*/
+#[no_mangle]
+pub extern "C" fn init(_ctx: FunctionCtx, _: FunctionArgs) -> c_int {
+ LUA_STATE
+ .try_with(|s| {
+ let l = s.clone().into_inner();
+ init_cluster_functions(l);
+ })
+ .unwrap();
+ 0
+}
+
#[derive(Serialize, Debug, Eq, PartialEq)]
pub struct QueryResult {
pub bucket_id: u64,
diff --git a/test_app/app/roles/api.lua b/test_app/app/roles/api.lua
index ae2cc904f9..1fbefe92ac 100644
--- a/test_app/app/roles/api.lua
+++ b/test_app/app/roles/api.lua
@@ -1,9 +1,10 @@
local vshard = require('vshard')
local cartridge = require('cartridge')
-local log = require('log')
+local yaml = require('yaml')
_G.query = nil
_G.insert_record = nil
+_G.sql_execute = nil
local function query(q)
local has_err, parser_res = pcall(
@@ -58,15 +59,25 @@ local function insert_record(space_name, values)
return res
end
+local function sql_execute(bucket_id, sql_q)
+ return yaml.decode(box.func["sbroad.execute_query"]:call({ bucket_id, sql_q }))
+end
+
local function init(opts) -- luacheck: no unused args
-- if opts.is_master then
-- end
_G.query = query
_G.insert_record = insert_record
+ _G.sql_execute = sql_execute
box.schema.func.create('sbroad.parse_sql', { if_not_exists = true, language = 'C' })
box.schema.func.create('sbroad.invalidate_caching_schema', { if_not_exists = true, language = 'C' })
box.schema.func.create('sbroad.calculate_bucket_id', { if_not_exists = true, language = 'C' })
+ box.schema.func.create('sbroad.execute_query', { if_not_exists = true, language = 'C' })
+ box.schema.func.create('sbroad.init', { if_not_exists = true, language = 'C' })
+
+ box.func["sbroad.init"]:call({})
+
return true
end
diff --git a/test_app/app/roles/storage.lua b/test_app/app/roles/storage.lua
index 31263a0bf2..53b5b8fa02 100644
--- a/test_app/app/roles/storage.lua
+++ b/test_app/app/roles/storage.lua
@@ -14,6 +14,11 @@ local function init(opts) -- luacheck: no unused args
-- if opts.is_master then
-- end
_G.insert_map = insert_map
+
+ box.schema.func.create('sbroad.init', { if_not_exists = true, language = 'C' })
+
+ box.func["sbroad.init"]:call({})
+
return true
end
diff --git a/test_app/test/integration/api_test.lua b/test_app/test/integration/api_test.lua
index 5f221d0ad8..ce2c178fa1 100644
--- a/test_app/test/integration/api_test.lua
+++ b/test_app/test/integration/api_test.lua
@@ -53,8 +53,6 @@ g.test_simple_query = function()
rows = {},
})
- -- fiber.sleep(300)
-
r, err = api:call("query", { [[SELECT * FROM "testing_space" where "id" = 1 and "name" = '123']] })
t.assert_equals(err, nil)
t.assert_equals(r, {
@@ -92,3 +90,22 @@ g.test_union_query = function()
},
})
end
+
+g.test_sql_execute = function ()
+ -- fiber.sleep(300)
+
+ local api = cluster:server("api-1").net_box
+ local r, err = api:call("sql_execute", { 359, [[SELECT * FROM "testing_space" where "id" = 1 and "name" = '123']] })
+ t.assert_equals(err, nil)
+ t.assert_equals(r, {
+ metadata = {
+ {name = "id", type = "integer"},
+ {name = "name", type = "string"},
+ {name = "product_units", type = "integer"},
+ {name = "bucket_id", type = "unsigned"},
+ },
+ rows = {
+ { 1, "123", 1, 359 }
+ },
+ })
+end
--
GitLab