box: introduce box_insert_arrow()

The new method inserts into a given space the data, provided in Arrow
columnar format [1]. At the moment it is not supported by memtx and vinyl
spaces.

1. https://arrow.apache.org/docs/format/Columnar.html

NO_DOC=Will be added along with the implementation for memtx
NO_CHANGELOG=See NO_DOC

Author: Gumix

extra/exports

@@ -73,6 +73,7 @@ box_index_tuple_position
 box_info_lsn
 box_init_latest_dd_version_id
 box_insert
+box_insert_arrow
 box_iproto_override
 box_iproto_send
 box_is_ro

src/box/box.cc

@@ -4011,6 +4011,19 @@ box_upsert(uint32_t space_id, uint32_t index_id, const char *tuple,
 	return box_process1(&request, result);
 }
 
+API_EXPORT int
+box_insert_arrow(uint32_t space_id, struct ArrowArray *array,
+		 struct ArrowSchema *schema)
+{
+	struct request request;
+	memset(&request, 0, sizeof(request));
+	request.type = IPROTO_INSERT_ARROW;
+	request.space_id = space_id;
+	request.arrow_array = array;
+	request.arrow_schema = schema;
+	return box_process1(&request, NULL);
+}
+
 /**
  * Trigger space truncation by bumping a counter
  * in _truncate space.

src/box/box.h

@@ -666,6 +666,31 @@ box_upsert(uint32_t space_id, uint32_t index_id, const char *tuple,
 	   const char *tuple_end, const char *ops, const char *ops_end,
 	   int index_base, box_tuple_t **result);
 
+struct ArrowArray;
+struct ArrowSchema;
+
+/**
+ * Executes a batch insert request.
+ *
+ * A record batch from the Arrow `array` is inserted into the space columns,
+ * whose names are provided by the Arrow `schema`. Column types in the schema
+ * must match the types of the corresponding fields in the space format.
+ *
+ * If a column is nullable in space format, it can be omitted. All non-nullable
+ * columns (including primary key parts) must be present in the batch.
+ *
+ * This function does not release neither `array` nor `schema`.
+ *
+ * \param space_id space identifier
+ * \param array input data in ArrowArray format
+ * \param schema definition of the input data in ArrowSchema format
+ * \retval 0 on success
+ * \retval -1 on error (check box_error_last())
+ */
+API_EXPORT int
+box_insert_arrow(uint32_t space_id, struct ArrowArray *array,
+		 struct ArrowSchema *schema);
+
 /**
  * Truncate space.
  *

src/box/space.c

@@ -1344,13 +1344,34 @@ after_old_tuple_lookup:;
 
 /**
  * Executes an IPROTO_INSERT_ARROW request. The request contains the data in
- * Arrow columnar format.
+ * Arrow columnar format that can appear in two guises:
+ *  1. in-memory data structures (arrow_array and arrow_schema);
+ *  2. serialized for interprocess communication (arrow_ipc and arrow_ipc_end).
  * Can return nonzero in case of error (diag is set).
  */
 static int
 space_execute_insert_arrow(struct space *space, struct txn *txn,
 			   struct request *request)
 {
+	if (request->arrow_array != NULL) {
+		assert(request->arrow_schema != NULL);
+		assert(request->arrow_ipc == NULL);
+		assert(request->arrow_ipc_end == NULL);
+		struct ArrowArray *array = request->arrow_array;
+		struct ArrowSchema *schema = request->arrow_schema;
+
+		if (space->vtab->execute_insert_arrow(
+				space, txn, array, schema) != 0)
+			return -1;
+		struct region *txn_region = tx_region_acquire(txn);
+		int rc = arrow_ipc_encode(
+				array, schema, txn_region, &request->arrow_ipc,
+				&request->arrow_ipc_end);
+		tx_region_release(txn, TX_ALLOC_SYSTEM);
+		return rc;
+	}
+
+	assert(request->arrow_schema == NULL);
 	assert(request->arrow_ipc != NULL);
 	assert(request->arrow_ipc_end != NULL);
 	struct ArrowArray array;

src/box/xrow.h

@@ -211,6 +211,10 @@ struct request {
 	/** Insert/replace/upsert tuple or proc argument or update operations. */
 	const char *tuple;
 	const char *tuple_end;
+	/** The data in in-memory Arrow format. */
+	struct ArrowArray *arrow_array;
+	/** Arrow schema for @arrow_array. */
+	struct ArrowSchema *arrow_schema;
 	/** The data in serialized Arrow format. */
 	const char *arrow_ipc;
 	/** End of @arrow_ipc. */

test/app-tap/module_api.c

@@ -15,6 +15,8 @@
 #include <lua.h>
 #include <lauxlib.h>
 
+#include "arrow/abi.h"
+
 #define STR2(x) #x
 #define STR(x) STR2(x)
 
@@ -3362,6 +3364,24 @@ test_fiber_basic_api(lua_State *L)
 	return 1;
 }
 
+static int
+test_box_insert_arrow(struct lua_State *L)
+{
+	fail_unless(lua_gettop(L) == 1);
+	fail_unless(lua_isnumber(L, 1));
+	uint32_t space_id = lua_tointeger(L, 1);
+	struct ArrowSchema schema;
+	struct ArrowArray array;
+	memset(&schema, 0, sizeof(schema));
+	memset(&array, 0, sizeof(array));
+
+	int rc = box_insert_arrow(space_id, &array, &schema);
+	fail_unless(rc == -1);
+	check_diag("ClientError", "memtx does not support arrow format");
+	lua_pushboolean(L, 1);
+	return 1;
+}
+
 LUA_API int
 luaopen_module_api(lua_State *L)
 {
@@ -3420,6 +3440,7 @@ luaopen_module_api(lua_State *L)
 		{"box_iproto_send", test_box_iproto_send},
 		{"box_iproto_override_set", test_box_iproto_override_set},
 		{"box_iproto_override_reset", test_box_iproto_override_reset},
+		{"box_insert_arrow", test_box_insert_arrow},
 		{NULL, NULL}
 	};
 	luaL_register(L, "module_api", lib);

test/app-tap/module_api.test.lua

@@ -670,8 +670,14 @@ local function test_box_ibuf(test, module)
     test:ok(module.box_ibuf(require('buffer').ibuf()), "box_ibuf API")
 end
 
+local function test_box_insert_arrow(test, module)
+    test:plan(1)
+    test:ok(module.box_insert_arrow(box.space.test.id),
+            "box_insert_arrow API")
+end
+
 require('tap').test("module_api", function(test)
-    test:plan(51)
+    test:plan(52)
     local status, module = pcall(require, 'module_api')
     test:is(status, true, "module")
     test:ok(status, "module is loaded")
@@ -709,6 +715,7 @@ require('tap').test("module_api", function(test)
     test:test("box_iproto_send", test_box_iproto_send, module)
     test:test("box_iproto_override", test_box_iproto_override, module)
     test:test("box_ibuf", test_box_ibuf, module)
+    test:test("box_insert_arrow", test_box_insert_arrow, module)
 
     space:drop()
 end)