box/lua: introduce space:insert_arrow(arrow)

The new method takes a string with the data in Arrow IPC format as an
argument, see commit be34a84 ("box: introduce box_insert_arrow()").
The data is deserialized and inserted into a space.

Follow-up tarantool#10508
Needed for tarantool/tarantool-ee#865

NO_DOC=No sense in mentioning in CE
NO_CHANGELOG=No sense in mentioning in CE

Author: Gumix

src/box/lua/index.c

@@ -37,6 +37,7 @@
 #include "box/lua/tuple.h"
 #include "box/lua/misc.h"
 #include "small/region.h"
+#include "arrow_ipc.h"
 #include "fiber.h"
 
 /** {{{ box.index Lua library: access to spaces and indexes
@@ -419,6 +420,33 @@ lbox_index_compact(lua_State *L)
 	return 0;
 }
 
+static int
+lbox_insert_arrow(lua_State *L)
+{
+	if (lua_gettop(L) != 2 || !lua_isnumber(L, 1) ||
+	    lua_type(L, 2) != LUA_TSTRING) {
+		diag_set(IllegalParams, "Usage: space:insert_arrow(arrow)");
+		return luaT_error(L);
+	}
+	uint32_t space_id = lua_tonumber(L, 1);
+
+	size_t len;
+	const char *data = lua_tolstring(L, 2, &len);
+	assert(data != NULL);
+	struct ArrowArray array;
+	struct ArrowSchema schema;
+	if (arrow_ipc_decode(&array, &schema, data, data + len) != 0)
+		return luaT_error(L);
+	int rc = box_insert_arrow(space_id, &array, &schema);
+	if (schema.release != NULL)
+		schema.release(&schema);
+	if (array.release != NULL)
+		array.release(&array);
+	if (rc != 0)
+		return luaT_error(L);
+	return 0;
+}
+
 /* }}} */
 
 void
@@ -452,6 +480,7 @@ box_lua_index_init(struct lua_State *L)
 		{"truncate", lbox_truncate},
 		{"stat", lbox_index_stat},
 		{"compact", lbox_index_compact},
+		{"insert_arrow", lbox_insert_arrow},
 		{NULL, NULL}
 	};
 

src/box/lua/schema.lua

@@ -2876,6 +2876,11 @@ space_mt.run_triggers = function(space, yesno)
     end
     builtin.space_run_triggers(s, yesno)
 end
+space_mt.insert_arrow = function(space, arrow)
+    check_space_arg(space, 'insert_arrow', 2)
+    check_space_exists(space, 2)
+    return internal.insert_arrow(space.id, arrow);
+end
 space_mt.frommap = box.internal.space.frommap
 space_mt.stat = box.internal.space.stat
 space_mt.__index = space_mt

src/lua/msgpack.c

@@ -519,6 +519,12 @@ luamp_decode_with_ctx(struct lua_State *L, struct luaL_serializer *cfg,
 			VERIFY(interval_unpack(data, len, itv) != NULL);
 			return;
 		}
+		case MP_ARROW:
+		{
+			lua_pushlstring(L, *data, len);
+			*data += len;
+			return;
+		}
 		default:
 			/* reset data to the extension header */
 			*data = svp;
@@ -531,7 +537,6 @@ luamp_decode_with_ctx(struct lua_State *L, struct luaL_serializer *cfg,
 	return;
 }
 
-
 static int
 lua_msgpack_encode(lua_State *L)
 {

test/box-luatest/gh_10508_space_insert_arrow_test.lua

@@ -0,0 +1,84 @@
+local server = require('luatest.server')
+local t = require('luatest')
+local g = t.group()
+
+g.before_all(function(cg)
+    cg.server = server:new()
+    cg.server:start()
+end)
+
+g.after_all(function(cg)
+    cg.server:drop()
+end)
+
+g.after_each(function(cg)
+    cg.server:exec(function()
+        if box.space.test ~= nil then
+            box.space.test:drop()
+        end
+    end)
+end)
+
+-- MP_EXT of type MP_ARROW, column 'a', value 0.
+local mp_arrow_hex = [[
+    c8011008ffffffff70000000040000009effffff0400010004000000b6ffffff0c0000000400
+    0000000000000100000004000000daffffff140000000202000004000000f0ffffff40000000
+    01000000610000000600080004000c0010000400080009000c000c000c000000040000000800
+    0a000c00040006000800ffffffff88000000040000008affffff040003001000000008000000
+    0000000000000000acffffff0100000000000000340000000800000000000000020000000000
+    0000000000000000000000000000000000000000000008000000000000000000000001000000
+    010000000000000000000000000000000a00140004000c0010000c0014000400060008000c00
+    00000000000000000000
+]]
+
+g.test_decode_mp_arrow = function(cg)
+    cg.server:exec(function(mp_arrow_hex)
+        local msgpack = require('msgpack')
+        -- Test invalid extension.
+        t.assert_error_msg_equals(
+            "Invalid MsgPack - invalid extension",
+            msgpack.decode, string.fromhex('c70408deadbeef'))
+        -- Test a valid extension.
+        local mp_arrow = string.fromhex(mp_arrow_hex:gsub("%s+", ""))
+        local decoded, next_pos = msgpack.decode(mp_arrow)
+        t.assert_equals(decoded, string.sub(mp_arrow, 5))
+        t.assert_equals(next_pos, 277)
+    end, {mp_arrow_hex})
+end
+
+g.test_space_insert_arrow = function(cg)
+    cg.server:exec(function(mp_arrow_hex)
+        local msgpack = require('msgpack')
+        local s = box.schema.create_space('test')
+
+        t.assert_error_msg_equals(
+            "Usage: space:insert_arrow(arrow)",
+            box.internal.insert_arrow)
+        t.assert_error_msg_equals(
+            "Usage: space:insert_arrow(arrow)",
+            box.internal.insert_arrow, 'not a number', 'string')
+        t.assert_error_msg_equals(
+            "Usage: space:insert_arrow(arrow)",
+            box.internal.insert_arrow, 11, {22})
+        t.assert_error_msg_equals(
+            "Use space:insert_arrow(...) instead of " ..
+            "space.insert_arrow(...)",
+            s.insert_arrow)
+        t.assert_error_msg_equals(
+            "Space 'xyz' does not exist",
+            s.insert_arrow, {id = 1000, name = 'xyz'})
+        t.assert_error_msg_equals(
+            "Usage: space:insert_arrow(arrow)",
+            s.insert_arrow, s, 22)
+        t.assert_error_msg_equals(
+            "Arrow decode error: Expected >= 544501618 bytes of remaining " ..
+            "data but found 29 bytes in buffer",
+            s.insert_arrow, s, 'not a valid arrow data string')
+
+        local mp_arrow = string.fromhex(mp_arrow_hex:gsub("%s+", ""))
+        local arrow = msgpack.decode(mp_arrow)
+        t.assert_error_msg_equals(
+            "memtx does not support arrow format",
+            s.insert_arrow, s, arrow)
+    end, {mp_arrow_hex})
+end