feat: migrate to lua table for executing query result

Was fixed interaction format between lua and sbroad rust lib. In current implementation sbroad encodes query result to yaml and send it to lua app. Lua app gives yaml and decodes it to lua table. After apply this changes lua app will receive lua table and yaml transformations will not be needed.

src/executor.rs

+pub mod result;

src/executor/result.rs

+use serde::ser::{Serialize, SerializeMap, Serializer};
+use tarantool::hlua::{self, LuaRead};
+
+use crate::ir::relation::Column;
+
+#[derive(LuaRead, Debug, PartialEq)]
+pub enum Value {
+    Boolean(bool),
+    Number(f64),
+    Integer(i64),
+    String(String),
+    Unsigned(u64),
+}
+
+/// Custom Implementation `ser::Serialize`, because if using standard `#derive[Serialize]` then each `Value` type
+/// record serialize as list and it doesn't correct for result format
+impl Serialize for Value {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match &self {
+            Value::Boolean(v) => serializer.serialize_bool(*v),
+            Value::Number(v) => serializer.serialize_f64(*v),
+            Value::Integer(v) => serializer.serialize_i64(*v),
+            Value::String(v) => serializer.serialize_str(v),
+            Value::Unsigned(v) => serializer.serialize_u64(*v),
+        }
+    }
+}
+
+impl Eq for Value {}
+
+#[derive(LuaRead, Debug, PartialEq, Eq)]
+pub struct BoxExecuteResult {
+    pub metadata: Vec<Column>,
+    pub rows: Vec<Vec<Value>>,
+}
+
+/// Custom Implementation `ser::Serialize`, because if using standard `#derive[Serialize]` then each `BoxExecuteResult`
+/// record is serialized to a list. That is not the result we expect.
+impl Serialize for BoxExecuteResult {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let mut map = serializer.serialize_map(Some(2))?;
+        map.serialize_entry("metadata", &self.metadata)?;
+        map.serialize_entry("rows", &self.rows)?;
+
+        map.end()
+    }
+}
+
+#[cfg(test)]
+mod tests;

src/executor/result/tests.rs

+use pretty_assertions::{assert_eq, assert_ne};
+
+use crate::ir::relation::Type;
+
+use super::*;
+
+#[test]
+fn box_execute_result_serialize() {
+    let r = BoxExecuteResult {
+        metadata: vec![
+            Column {
+                name: "id".into(),
+                r#type: Type::Integer,
+            },
+            Column {
+                name: "name".into(),
+                r#type: Type::String,
+            },
+            Column {
+                name: "count".into(),
+                r#type: Type::Unsigned,
+            },
+        ],
+        rows: vec![vec![
+            Value::Integer(1),
+            Value::String("тест".into()),
+            Value::Unsigned(1),
+        ]],
+    };
+
+    let actual = rmp_serde::to_vec(&r).unwrap();
+
+    // Incorrect serialize message with default msgpack serializer
+    let default_serialize_msg = vec![
+        0x92, 0x93, 0x82, 0xA4, 0x6E, 0x61, 0x6D, 0x65, 0xA2, 0x69, 0x64, 0xA4, 0x74, 0x79, 0x70,
+        0x65, 0xA7, 0x69, 0x6E, 0x74, 0x65, 0x67, 0x65, 0x72, 0x82, 0xA4, 0x6E, 0x61, 0x6D, 0x65,
+        0xA4, 0x6E, 0x61, 0x6D, 0x65, 0xA4, 0x74, 0x79, 0x70, 0x65, 0xA6, 0x73, 0x74, 0x72, 0x69,
+        0x6E, 0x67, 0x82, 0xA4, 0x6E, 0x61, 0x6D, 0x65, 0xA5, 0x63, 0x6F, 0x75, 0x6E, 0x74, 0xA4,
+        0x74, 0x79, 0x70, 0x65, 0xA8, 0x75, 0x6E, 0x73, 0x69, 0x67, 0x6E, 0x65, 0x64, 0x91, 0x93,
+        0x81, 0x2, 0x1, 0x81, 0x3, 0xA8, 0xD1, 0x82, 0xD0, 0xB5, 0xD1, 0x81, 0xD1, 0x82, 0x81, 0x4,
+        0x1,
+    ];
+    assert_ne!(default_serialize_msg, actual);
+
+    // Expected value was got in debugging interaction lua app and current rust lib
+    let correct_serialize_msg_expected = vec![
+        0x82, 0xA8, 0x6D, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x93, 0x82, 0xA4, 0x6E, 0x61,
+        0x6D, 0x65, 0xA2, 0x69, 0x64, 0xA4, 0x74, 0x79, 0x70, 0x65, 0xA7, 0x69, 0x6E, 0x74, 0x65,
+        0x67, 0x65, 0x72, 0x82, 0xA4, 0x6E, 0x61, 0x6D, 0x65, 0xA4, 0x6E, 0x61, 0x6D, 0x65, 0xA4,
+        0x74, 0x79, 0x70, 0x65, 0xA6, 0x73, 0x74, 0x72, 0x69, 0x6E, 0x67, 0x82, 0xA4, 0x6E, 0x61,
+        0x6D, 0x65, 0xA5, 0x63, 0x6F, 0x75, 0x6E, 0x74, 0xA4, 0x74, 0x79, 0x70, 0x65, 0xA8, 0x75,
+        0x6E, 0x73, 0x69, 0x67, 0x6E, 0x65, 0x64, 0xA4, 0x72, 0x6F, 0x77, 0x73, 0x91, 0x93, 0x1,
+        0xA8, 0xD1, 0x82, 0xD0, 0xB5, 0xD1, 0x81, 0xD1, 0x82, 0x1,
+    ];
+    assert_eq!(correct_serialize_msg_expected, actual);
+}

src/lib.rs

 //! Tarantool planner for distributed SQL.
 mod bucket;
 mod errors;
+mod executor;
 pub mod ir;
 mod lua_bridge;
 mod parser;

src/lua_bridge.rs

+use crate::executor::result::BoxExecuteResult;
 use tarantool::ffi::tarantool::luaT_state;
 use tarantool::hlua::{Lua, LuaError, LuaFunction};
+use tarantool::log::{say, SayLevel};
 
+/// Function get cartridge cluster schema
 pub fn get_cluster_schema() -> Result<String, LuaError> {
     let lua = unsafe { Lua::from_existing_state(luaT_state(), false) };
 
@@ -10,7 +13,8 @@ pub fn get_cluster_schema() -> Result<String, LuaError> {
     Ok(res)
 }
 
-pub fn exec_query(bucket_id: u64, query: &str) -> Result<String, LuaError> {
+/// Function execute sql query on selected node
+pub fn exec_query(bucket_id: u64, query: &str) -> Result<BoxExecuteResult, LuaError> {
     let lua = unsafe { Lua::from_existing_state(luaT_state(), false) };
 
     lua.exec(
@@ -19,7 +23,7 @@ pub fn exec_query(bucket_id: u64, query: &str) -> Result<String, LuaError> {
         local yaml = require('yaml')
 
         function execute_sql(bucket_id, query)
-            local data, err = vshard.router.call(
+            local res, err = vshard.router.call(
                 bucket_id,
                 'read',
                 'box.execute',
@@ -30,13 +34,21 @@ pub fn exec_query(bucket_id: u64, query: &str) -> Result<String, LuaError> {
                 error(err)
             end
 
-            return yaml.encode(data)
+            return res
         end
     "#,
     )?;
 
     let exec_sql: LuaFunction<_> = lua.get("execute_sql").unwrap();
-    let res = exec_sql.call_with_args((bucket_id, query))?;
+    let res: BoxExecuteResult = exec_sql.call_with_args((bucket_id, query))?;
+
+    say(
+        SayLevel::Error,
+        "lua_bridge.rs",
+        110,
+        None,
+        &format!("{:?}", res),
+    );
 
     Ok(res)
 }

test_app/app/roles/api.lua

 local vshard = require('vshard')
 local cartridge = require('cartridge')
-local yaml = require('yaml')
 
 _G.query = nil
 _G.insert_record = nil
@@ -19,7 +18,7 @@ local function query(q)
 
     local result = nil
     for _, rec in pairs(parser_res) do
-        local res = yaml.decode(box.func["sbroad.execute_query"]:call({ rec[1], rec[2] }))
+        local res = box.func["sbroad.execute_query"]:call({ rec[1], rec[2] })
 
         if result == nil then
             result = res