box: release array and schema in box_insert_arrow

According to the documentation to the Apache Arrow C data interface:

> Consumers MUST call a base structure’s release callback when they
> won’t be using it anymore[^1]

`box_insert_arrow()` is clearly a consumer so it must move or release
the input Arrow array and schema to conform to the specification while
currently it does not. This patch fixes this issue.

Note that for performance reasons (to avoid extra copying), the engine
callback that implements batch insertion should be allowed to move the
input Arrow array and schema to a private location. To allow for that,
we have to move the code that encodes the input Arrow in the IPC format
for xlog from `txn_add_redo()` to `space_execute_insert_arrow()`.

Closes tarantool#11338

[^1]: https://arrow.apache.org/docs/format/CDataInterface.html#release-callback-semantics-for-consumers

NO_DOC=bug fix
NO_CHANGELOG=batch insertion isn't implemented in CE

Author: locker

perf/lua/column_insert_module.c

@@ -276,8 +276,6 @@ insert_batch_lua_func(struct lua_State *L)
 				 batch_row_count, sparse_mode);
 		if (box_insert_arrow(space_id, &array, &schema) != 0)
 			return luaT_error(L);
-		schema.release(&schema);
-		array.release(&array);
 	}
 	return 0;
 }

src/box/box.cc

@@ -4134,7 +4134,12 @@ box_insert_arrow(uint32_t space_id, struct ArrowArray *array,
 	request.space_id = space_id;
 	request.arrow_array = array;
 	request.arrow_schema = schema;
-	return box_process1(&request, NULL);
+	int rc = box_process1(&request, NULL);
+	if (array->release != NULL)
+		array->release(array);
+	if (schema->release != NULL)
+		schema->release(schema);
+	return rc;
 }
 
 /**

src/box/box.h

@@ -683,7 +683,7 @@ struct ArrowSchema;
  * 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`.
+ * Both `array` and `schema` are moved or released, even on failure.
  *
  * \param space_id space identifier
  * \param array input data in ArrowArray format

src/box/lua/index.c

@@ -437,12 +437,7 @@ lbox_insert_arrow(lua_State *L)
 	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)
+	if (box_insert_arrow(space_id, &array, &schema) != 0)
 		return luaT_error(L);
 	return 0;
 }

src/box/space.c

@@ -1363,16 +1363,26 @@ space_execute_insert_arrow(struct space *space, struct txn *txn,
 	int rc;
 	struct region *gc = &fiber()->gc;
 	size_t gc_svp = region_used(gc);
-	bool do_ipc_decode = request->arrow_ipc != NULL;
 	struct ArrowArray *array = request->arrow_array;
 	struct ArrowSchema *schema = request->arrow_schema;
 
-	if (do_ipc_decode) {
+	if (request->arrow_ipc == NULL) {
+		assert(array != NULL);
+		assert(schema != NULL);
+		assert(request->arrow_ipc_end == NULL);
+		struct region *txn_region = tx_region_acquire(txn);
+		rc = arrow_ipc_encode(array, schema, txn_region,
+				      &request->arrow_ipc,
+				      &request->arrow_ipc_end);
+		tx_region_release(txn, TX_ALLOC_SYSTEM);
+		if (rc != 0)
+			goto eof;
+	} else {
 		assert(array == NULL);
 		assert(schema == NULL);
 		assert(request->arrow_ipc_end != NULL);
-		array = xregion_alloc_object(&fiber()->gc, struct ArrowArray);
-		schema = xregion_alloc_object(&fiber()->gc, struct ArrowSchema);
+		array = xregion_alloc_object(gc, struct ArrowArray);
+		schema = xregion_alloc_object(gc, struct ArrowSchema);
 		rc = arrow_ipc_decode(array, schema, request->arrow_ipc,
 				      request->arrow_ipc_end);
 		if (rc != 0)
@@ -1381,12 +1391,10 @@ space_execute_insert_arrow(struct space *space, struct txn *txn,
 
 	rc = space->vtab->execute_insert_arrow(space, txn, array, schema);
 
-	if (do_ipc_decode) {
-		assert(array->release != NULL);
-		assert(schema->release != NULL);
+	if (array->release != NULL)
 		array->release(array);
+	if (schema->release != NULL)
 		schema->release(schema);
-	}
 eof:
 	region_truncate(gc, gc_svp);
 	return rc;

src/box/space.h

@@ -71,6 +71,11 @@ struct space_vtab {
 	int (*execute_update)(struct space *, struct txn *,
 			      struct request *, struct tuple **result);
 	int (*execute_upsert)(struct space *, struct txn *, struct request *);
+	/**
+	 * Executes a batch insert request.
+	 *
+	 * The implementation may move or release the input array and schema.
+	 */
 	int (*execute_insert_arrow)(struct space *space, struct txn *txn,
 				    struct ArrowArray *array,
 				    struct ArrowSchema *schema);

src/box/txn.c

@@ -42,7 +42,6 @@
 #include "session.h"
 #include "wal_ext.h"
 #include "rmean.h"
-#include "arrow_ipc.h"
 
 double too_long_threshold;
 
@@ -280,21 +279,8 @@ txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 	if (space != NULL && space->wal_ext != NULL)
 		space_wal_ext_process_request(space->wal_ext, stmt, request);
 	struct region *txn_region = tx_region_acquire(txn);
-	if (request->arrow_array != NULL) {
-		assert(request->arrow_schema != NULL);
-		assert(request->arrow_ipc == NULL);
-		assert(request->arrow_ipc_end == NULL);
-		if (arrow_ipc_encode(
-				request->arrow_array, request->arrow_schema,
-				txn_region, &request->arrow_ipc,
-				&request->arrow_ipc_end) != 0) {
-			tx_region_release(txn, TX_ALLOC_SYSTEM);
-			return -1;
-		}
-	}
 	xrow_encode_dml(request, txn_region, row->body, &row->bodycnt);
 	tx_region_release(txn, TX_ALLOC_SYSTEM);
-	txn_region = NULL;
 	stmt->row = row;
 	return 0;
 }

test/app-tap/module_api.c

@@ -3364,19 +3364,41 @@ test_fiber_basic_api(lua_State *L)
 	return 1;
 }
 
+static void
+arrow_schema_destroy(struct ArrowSchema *schema)
+{
+	schema->release = NULL;
+}
+
+static void
+arrow_array_destroy(struct ArrowArray *array)
+{
+	array->release = NULL;
+}
+
 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));
+	schema.format = "+s";
+	schema.release = arrow_schema_destroy;
+
+	struct ArrowArray array;
 	memset(&array, 0, sizeof(array));
+	array.n_buffers = 1;
+	const void *buffers[1] = {NULL};
+	array.buffers = buffers;
+	array.release = arrow_array_destroy;
 
 	int rc = box_insert_arrow(space_id, &array, &schema);
 	fail_unless(rc == -1);
+	fail_unless(array.release == NULL);
+	fail_unless(schema.release == NULL);
 	check_diag("ClientError", "memtx does not support arrow format");
 	lua_pushboolean(L, 1);
 	return 1;