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.

Also the IPROTO_INSERT_ARROW request is introduced. It inserts the data
that is serialized into Arrow IPC format [2].

1. https://arrow.apache.org/docs/format/Columnar.html
2. https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc

Closes #10508
Needed for tarantool/tarantool-ee#820

NO_CHANGELOG=No sense in mentioning in CE release notes

@TarantoolBot document
Title: Document iproto request IPROTO_INSERT_ARROW
Product: Tarantool
Since: 3.3
Root documents: https://www.tarantool.io/en/doc/latest/reference/internals/iproto/requests/
https://www.tarantool.io/en/doc/latest/reference/internals/iproto/keys/#internals-iproto-keys-features

Available IPROTO_FEATURES are the following:
* IPROTO_FEATURE_INSERT_ARROW = 12 - Support of data insertion in Arrow
  format.
IPROTO version supporting the this feature is 10 or newer.

Client-server requests and responses:
* IPROTO_INSERT_ARROW = 0x11 - Insert Arrow data request.

The structure of the `IPROTO_INSERT_ARROW` request is similar to
`IPROTO_INSERT`, the only difference is `IPROTO_ARROW : MP_ARROW`
instead of `IPROTO_TUPLE : MP_ARRAY`.

The response to `IPROTO_INSERT_ARROW` contains an empty body.

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/CMakeLists.txt

@@ -282,6 +282,7 @@ set(box_sources
     decimal.c
     read_view.c
     mp_box_ctx.c
+    arrow_ipc.c
     ${sql_sources}
     ${lua_sources}
     lua/init.c

src/box/arrow_ipc.c

@@ -0,0 +1,166 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2024, Tarantool AUTHORS, please see AUTHORS file.
+ */
+#include "arrow_ipc.h"
+
+#include "diag.h"
+#include "error.h"
+#include "small/region.h"
+#include "nanoarrow/nanoarrow_ipc.h"
+
+int
+arrow_ipc_encode(struct ArrowArray *array, struct ArrowSchema *schema,
+		 struct region *region, const char **ret_data,
+		 const char **ret_data_end)
+{
+	ArrowErrorCode rc;
+	struct ArrowError error;
+	struct ArrowBuffer buffer;
+	ArrowBufferInit(&buffer);
+
+	struct ArrowArrayView array_view;
+	rc = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
+			 "ArrowArrayViewInitFromSchema", error.message);
+		return -1;
+	}
+
+	/* Set buffer sizes and data pointers from an array. */
+	rc = ArrowArrayViewSetArray(&array_view, array, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
+			 "ArrowArrayViewSetArray", error.message);
+		goto error1;
+	}
+
+	/* All bytes written to the stream will be appended to the buffer. */
+	struct ArrowIpcOutputStream stream;
+	rc = ArrowIpcOutputStreamInitBuffer(&stream, &buffer);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
+			 "ArrowIpcOutputStreamInitBuffer", NULL);
+		goto error1;
+	}
+
+	/*
+	 * A stream writer which encodes schema and array into an IPC byte
+	 * stream. The writer takes ownership of the output byte stream.
+	 */
+	struct ArrowIpcWriter writer;
+	rc = ArrowIpcWriterInit(&writer, &stream);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE, "ArrowIpcWriterInit",
+			 NULL);
+		stream.release(&stream);
+		goto error1;
+	}
+
+	rc = ArrowIpcWriterWriteSchema(&writer, schema, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
+			 "ArrowIpcWriterWriteSchema", error.message);
+		goto error2;
+	}
+
+	rc = ArrowIpcWriterWriteArrayView(&writer, &array_view, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
+			 "ArrowIpcWriterWriteArrayView", error.message);
+		goto error2;
+	}
+
+	/*
+	 * TODO: It is possible to avoid extra `memcpy()' by switching
+	 * `ArrowBuffer' to `region_realloc()'.
+	 */
+	char *data = xregion_alloc(region, buffer.size_bytes);
+	memcpy(data, buffer.data, buffer.size_bytes);
+	*ret_data = data;
+	*ret_data_end = data + buffer.size_bytes;
+
+	ArrowIpcWriterReset(&writer);
+	ArrowArrayViewReset(&array_view);
+	ArrowBufferReset(&buffer);
+	return 0;
+error2:
+	ArrowIpcWriterReset(&writer);
+error1:
+	ArrowArrayViewReset(&array_view);
+	ArrowBufferReset(&buffer);
+	return -1;
+}
+
+int
+arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
+		 const char *data, const char *data_end)
+{
+	ssize_t size = data_end - data;
+	if (size <= 0) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE, NULL,
+			 "Unexpected data size");
+		return -1;
+	}
+
+	ArrowErrorCode rc;
+	struct ArrowError error;
+	struct ArrowBuffer buffer;
+	ArrowBufferInit(&buffer);
+
+	rc = ArrowBufferAppend(&buffer, data, size);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE, "ArrowBufferAppend",
+			 NULL);
+		ArrowBufferReset(&buffer);
+		return -1;
+	}
+
+	/*
+	 * Create an input stream from a buffer.
+	 * The stream takes ownership of the buffer and reads bytes from it.
+	 */
+	struct ArrowIpcInputStream input_stream;
+	rc = ArrowIpcInputStreamInitBuffer(&input_stream, &buffer);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE,
+			 "ArrowIpcInputStreamInitBuffer", NULL);
+		ArrowBufferReset(&buffer);
+		return -1;
+	}
+
+	/*
+	 * Initialize an array stream from an input stream of bytes.
+	 * The array_stream takes ownership of input_stream.
+	 */
+	struct ArrowArrayStream array_stream;
+	rc = ArrowIpcArrayStreamReaderInit(&array_stream, &input_stream, NULL);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE,
+			 "ArrowIpcArrayStreamReaderInit", NULL);
+		input_stream.release(&input_stream);
+		return -1;
+	}
+
+	rc = ArrowArrayStreamGetSchema(&array_stream, schema, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE,
+			 "ArrowArrayStreamGetSchema", error.message);
+		goto error;
+	}
+
+	rc = ArrowArrayStreamGetNext(&array_stream, array, &error);
+	if (rc != NANOARROW_OK) {
+		diag_set(ClientError, ER_ARROW_IPC_DECODE,
+			 "ArrowArrayStreamGetNext", error.message);
+		schema->release(schema);
+		goto error;
+	}
+
+	ArrowArrayStreamRelease(&array_stream);
+	return 0;
+error:
+	ArrowArrayStreamRelease(&array_stream);
+	return -1;
+}

src/box/arrow_ipc.h

@@ -0,0 +1,36 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2024, Tarantool AUTHORS, please see AUTHORS file.
+ */
+#pragma once
+
+#include "arrow/abi.h"
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+struct region;
+
+/**
+ * Encodes `array' and `schema' into Arrow IPC format. The memory is allocated
+ * on `region', and the address is returned via `ret_data' and `ret_data_end'.
+ * Returns 0 on success, -1 on failure (diag is set).
+ */
+int
+arrow_ipc_encode(struct ArrowArray *array, struct ArrowSchema *schema,
+		 struct region *region, const char **ret_data,
+		 const char **ret_data_end);
+
+/**
+ * Decodes `array' and `schema' from the `data' in Arrow IPC format.
+ * Returns 0 on success, -1 on failure (diag is set).
+ */
+int
+arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
+		 const char *data, const char *data_end);
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */

src/box/blackhole.c

@@ -113,6 +113,7 @@ static const struct space_vtab blackhole_space_vtab = {
 	/* .execute_delete = */ blackhole_space_execute_delete,
 	/* .execute_update = */ blackhole_space_execute_update,
 	/* .execute_upsert = */ blackhole_space_execute_upsert,
+	/* .execute_insert_arrow = */ generic_space_execute_insert_arrow,
 	/* .ephemeral_replace = */ generic_space_ephemeral_replace,
 	/* .ephemeral_delete = */ generic_space_ephemeral_delete,
 	/* .ephemeral_rowid_next = */ generic_space_ephemeral_rowid_next,

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/errcode.h

@@ -433,6 +433,8 @@ struct errcode_record {
 	_(ER_READ_VIEW_CLOSED, 286,		"The read view is closed") \
 	_(ER_WAL_QUEUE_FULL, 287,		"The WAL queue is full") \
 	_(ER_INVALID_VCLOCK, 288,		"Invalid vclock", "value", STRING) \
+	_(ER_ARROW_IPC_ENCODE, 289,		"Failed to encode Arrow IPC data", "method", STRING, "details", STRING) \
+	_(ER_ARROW_IPC_DECODE, 290,		"Failed to decode Arrow IPC data", "method", STRING, "details", STRING) \
 	TEST_ERROR_CODES(_) /** This one should be last. */
 
 /*

src/box/iproto.cc

@@ -1866,6 +1866,7 @@ iproto_msg_decode(struct iproto_msg *msg, struct cmsg_hop **route)
 	case IPROTO_UPDATE:
 	case IPROTO_DELETE:
 	case IPROTO_UPSERT:
+	case IPROTO_INSERT_ARROW:
 		assert(type < sizeof(iproto_thread->dml_route) /
 			      sizeof(*iproto_thread->dml_route));
 		*route = iproto_thread->dml_route[type];
@@ -3655,6 +3656,7 @@ iproto_thread_init_routes(struct iproto_thread *iproto_thread)
 	assert(dml_route[IPROTO_BEGIN] == NULL);
 	assert(dml_route[IPROTO_COMMIT] == NULL);
 	assert(dml_route[IPROTO_ROLLBACK] == NULL);
+	dml_route[IPROTO_INSERT_ARROW] = iproto_thread->process1_route;
 
 	iproto_thread->connect_route[0] =
 		{ tx_process_connect, &iproto_thread->net_pipe };

src/box/iproto_constants.c

@@ -32,23 +32,24 @@
 
 #define bit(c) (1ULL<<IPROTO_##c)
 const uint64_t iproto_body_key_map[IPROTO_TYPE_STAT_MAX] = {
-	0,                                                     /* unused */
-	bit(SPACE_ID) | bit(LIMIT) | bit(KEY),                 /* SELECT */
-	bit(SPACE_ID) | bit(TUPLE),                            /* INSERT */
-	bit(SPACE_ID) | bit(TUPLE),                            /* REPLACE */
-	bit(SPACE_ID) | bit(KEY) | bit(TUPLE),                 /* UPDATE */
-	bit(SPACE_ID) | bit(KEY),                              /* DELETE */
-	0,                                                     /* CALL_16 */
-	0,                                                     /* AUTH */
-	0,                                                     /* EVAL */
-	bit(SPACE_ID) | bit(OPS) | bit(TUPLE),                 /* UPSERT */
-	0,                                                     /* CALL */
-	0,                                                     /* EXECUTE */
-	0,                                                     /* NOP */
-	0,                                                     /* PREPARE */
-	0,                                                     /* BEGIN */
-	0,                                                     /* COMMIT */
-	0,                                                     /* ROLLBACK */
+	0,                                                    /* unused */
+	bit(SPACE_ID) | bit(LIMIT) | bit(KEY),                /* SELECT */
+	bit(SPACE_ID) | bit(TUPLE),                           /* INSERT */
+	bit(SPACE_ID) | bit(TUPLE),                           /* REPLACE */
+	bit(SPACE_ID) | bit(KEY) | bit(TUPLE),                /* UPDATE */
+	bit(SPACE_ID) | bit(KEY),                             /* DELETE */
+	0,                                                    /* CALL_16 */
+	0,                                                    /* AUTH */
+	0,                                                    /* EVAL */
+	bit(SPACE_ID) | bit(OPS) | bit(TUPLE),                /* UPSERT */
+	0,                                                    /* CALL */
+	0,                                                    /* EXECUTE */
+	0,                                                    /* NOP */
+	0,                                                    /* PREPARE */
+	0,                                                    /* BEGIN */
+	0,                                                    /* COMMIT */
+	0,                                                    /* ROLLBACK */
+	bit(SPACE_ID) | bit(ARROW),                           /* INSERT_ARROW */
 };
 #undef bit