box: move arrow_ipc.[hc] from src/box/ to src/lib/core/

Arrow format serialization is actually box-independent and suits better for
the `core` directory. However the ClientError can not be set outside of the
`box`, that's why the EncodeError and DecodeError were introduced.

Follow-up tarantool#10508

NO_DOC=refactoring
NO_CHANGELOG=refactoring

Author: Gumix

src/box/CMakeLists.txt

@@ -134,7 +134,6 @@ add_custom_target(box_generate_lua_sources
 set_property(DIRECTORY PROPERTY ADDITIONAL_MAKE_CLEAN_FILES ${lua_sources})
 
 include_directories(${ZSTD_INCLUDE_DIRS})
-include_directories(${NANOARROW_INCLUDE_DIRS})
 include_directories(${PROJECT_BINARY_DIR}/src/box/sql)
 include_directories(${PROJECT_BINARY_DIR}/src/box)
 include_directories(${EXTRA_CORE_INCLUDE_DIRS})
@@ -283,7 +282,6 @@ set(box_sources
     decimal.c
     read_view.c
     mp_box_ctx.c
-    arrow_ipc.c
     ${sql_sources}
     ${lua_sources}
     lua/init.c

src/box/errcode.h

@@ -433,9 +433,7 @@ 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) \
-	_(ER_SYNC_QUEUE_FULL, 291,		"The synchronous transaction queue is full") \
+	_(ER_SYNC_QUEUE_FULL, 289,		"The synchronous transaction queue is full") \
 	TEST_ERROR_CODES(_) /** This one should be last. */
 
 /*

src/box/mp_error.cc

@@ -259,6 +259,10 @@ error_build(struct mp_error *mp_error)
 		err = new SSLError();
 	} else if (strcmp(mp_error->type, "FileFormatError") == 0) {
 		err = new FileFormatError();
+	} else if (strcmp(mp_error->type, "EncodeError") == 0) {
+		err = new EncodeError();
+	} else if (strcmp(mp_error->type, "DecodeError") == 0) {
+		err = new DecodeError();
 	} else {
 		err = new ClientError();
 	}

src/lib/core/CMakeLists.txt

@@ -28,6 +28,7 @@ set(core_sources
     random.c
     trigger.cc
     port.c
+    arrow_ipc.c
     decimal.c
     mp_decimal.c
     cord_buf.c
@@ -67,6 +68,7 @@ else()
 endif()
 
 include_directories(${OPENSSL_INCLUDE_DIR}
+                    ${NANOARROW_INCLUDE_DIRS}
                     ${EXTRA_CORE_INCLUDE_DIRS})
 
 if (TARGET_OS_NETBSD)

src/box/arrow_ipc.c renamed to src/lib/core/arrow_ipc.c

@@ -6,7 +6,6 @@
 #include "arrow_ipc.h"
 
 #include "diag.h"
-#include "error.h"
 #include "small/region.h"
 #include "nanoarrow/nanoarrow_ipc.h"
 
@@ -23,25 +22,22 @@ arrow_ipc_encode(struct ArrowArray *array, struct ArrowSchema *schema,
 	struct ArrowArrayView array_view;
 	rc = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
-			 "ArrowArrayViewInitFromSchema", error.message);
+		diag_set(EncodeError, "Arrow", 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);
+		diag_set(EncodeError, "Arrow", 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);
+		diag_set(EncodeError, "Arrow", tt_strerror(rc));
 		goto error1;
 	}
 
@@ -52,23 +48,20 @@ arrow_ipc_encode(struct ArrowArray *array, struct ArrowSchema *schema,
 	struct ArrowIpcWriter writer;
 	rc = ArrowIpcWriterInit(&writer, &stream);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_ENCODE, "ArrowIpcWriterInit",
-			 NULL);
+		diag_set(EncodeError, "Arrow", tt_strerror(rc));
 		stream.release(&stream);
 		goto error1;
 	}
 
 	rc = ArrowIpcWriterWriteSchema(&writer, schema, &error);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
-			 "ArrowIpcWriterWriteSchema", error.message);
+		diag_set(EncodeError, "Arrow", error.message);
 		goto error2;
 	}
 
 	rc = ArrowIpcWriterWriteArrayView(&writer, &array_view, &error);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_ENCODE,
-			 "ArrowIpcWriterWriteArrayView", error.message);
+		diag_set(EncodeError, "Arrow", error.message);
 		goto error2;
 	}
 
@@ -99,8 +92,7 @@ arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
 {
 	ssize_t size = data_end - data;
 	if (size <= 0) {
-		diag_set(ClientError, ER_ARROW_IPC_DECODE, NULL,
-			 "Unexpected data size");
+		diag_set(DecodeError, "Arrow", "unexpected data size");
 		return -1;
 	}
 
@@ -111,8 +103,7 @@ arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
 
 	rc = ArrowBufferAppend(&buffer, data, size);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_DECODE, "ArrowBufferAppend",
-			 NULL);
+		diag_set(DecodeError, "Arrow", tt_strerror(rc));
 		ArrowBufferReset(&buffer);
 		return -1;
 	}
@@ -124,8 +115,7 @@ arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
 	struct ArrowIpcInputStream input_stream;
 	rc = ArrowIpcInputStreamInitBuffer(&input_stream, &buffer);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_DECODE,
-			 "ArrowIpcInputStreamInitBuffer", NULL);
+		diag_set(DecodeError, "Arrow", tt_strerror(rc));
 		ArrowBufferReset(&buffer);
 		return -1;
 	}
@@ -137,23 +127,20 @@ arrow_ipc_decode(struct ArrowArray *array, struct ArrowSchema *schema,
 	struct ArrowArrayStream array_stream;
 	rc = ArrowIpcArrayStreamReaderInit(&array_stream, &input_stream, NULL);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_DECODE,
-			 "ArrowIpcArrayStreamReaderInit", NULL);
+		diag_set(DecodeError, "Arrow", tt_strerror(rc));
 		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);
+		diag_set(DecodeError, "Arrow", error.message);
 		goto error;
 	}
 
 	rc = ArrowArrayStreamGetNext(&array_stream, array, &error);
 	if (rc != NANOARROW_OK) {
-		diag_set(ClientError, ER_ARROW_IPC_DECODE,
-			 "ArrowArrayStreamGetNext", error.message);
+		diag_set(DecodeError, "Arrow", error.message);
 		schema->release(schema);
 		goto error;
 	}

src/box/arrow_ipc.h renamed to src/lib/core/arrow_ipc.h

@@ -439,6 +439,12 @@ struct error *
 BuildCryptoError(const char *file, unsigned line, const char *format, ...);
 struct error *
 BuildRaftError(const char *file, unsigned line, const char *format, ...);
+struct error *
+BuildEncodeError(const char *file, unsigned line, const char *format,
+		 const char *details);
+struct error *
+BuildDecodeError(const char *file, unsigned line, const char *format,
+		 const char *details);
 
 /**
  * Allocate and create new FileFormatError. In case of OOM return OutOfMemory.

src/lib/core/diag.h

@@ -296,6 +296,32 @@ FileFormatError::FileFormatError(const char *file, unsigned line,
 	va_end(ap);
 }
 
+const struct type_info type_EncodeError =
+	make_type("EncodeError", &type_Exception);
+
+EncodeError::EncodeError(const char *file, unsigned line, const char *format,
+			 ...)
+	: Exception(&type_EncodeError, file, line)
+{
+	va_list ap;
+	va_start(ap, format);
+	error_vformat_msg(this, format, ap);
+	va_end(ap);
+}
+
+const struct type_info type_DecodeError =
+	make_type("DecodeError", &type_Exception);
+
+DecodeError::DecodeError(const char *file, unsigned line, const char *format,
+			 ...)
+	: Exception(&type_DecodeError, file, line)
+{
+	va_list ap;
+	va_start(ap, format);
+	error_vformat_msg(this, format, ap);
+	va_end(ap);
+}
+
 struct error *
 BuildOutOfMemory(const char *file, unsigned line,
 		 size_t amount, const char *allocator,
@@ -425,3 +451,25 @@ BuildFileFormatError(const char *file, unsigned line, const char *format, ...)
 	va_end(ap);
 	return e;
 }
+
+struct error *
+BuildEncodeError(const char *file, unsigned line, const char *format,
+		 const char *details)
+{
+	EncodeError *e = new EncodeError(file, line, "%s encode error: %s",
+					 format, details);
+	error_payload_set_str(&e->payload, "format", format);
+	error_payload_set_str(&e->payload, "details", details);
+	return e;
+}
+
+struct error *
+BuildDecodeError(const char *file, unsigned line, const char *format,
+		 const char *details)
+{
+	DecodeError *e = new DecodeError(file, line, "%s decode error: %s",
+					 format, details);
+	error_payload_set_str(&e->payload, "format", format);
+	error_payload_set_str(&e->payload, "details", details);
+	return e;
+}

src/lib/core/exception.cc

@@ -57,6 +57,10 @@ extern const struct type_info type_RaftError;
 extern const struct type_info type_FileFormatError;
 /* type_info for FiberSliceIsExceeded exception */
 extern const struct type_info type_FiberSliceIsExceeded;
+/* type_info for EncodeError exception */
+extern const struct type_info type_EncodeError;
+/* type_info for DecodeError exception */
+extern const struct type_info type_DecodeError;
 
 const char *
 exception_get_string(struct error *e, const struct method_info *method);
@@ -271,6 +275,28 @@ class FileFormatError: public Exception {
 	virtual void raise() { throw this; }
 };
 
+class EncodeError: public Exception {
+public:
+	EncodeError(const char *file, unsigned line, const char *format, ...);
+
+	EncodeError()
+		: Exception(&type_EncodeError, NULL, 0)
+	{
+	}
+	virtual void raise() { throw this; }
+};
+
+class DecodeError: public Exception {
+public:
+	DecodeError(const char *file, unsigned line, const char *format, ...);
+
+	DecodeError()
+		: Exception(&type_DecodeError, NULL, 0)
+	{
+	}
+	virtual void raise() { throw this; }
+};
+
 #define tnt_error(class, ...) ({					\
 	say_debug("%s at %s:%i", #class, __FILE__, __LINE__);		\
 	class *e = new class(__FILE__, __LINE__, ##__VA_ARGS__);	\

src/lib/core/exception.h

@@ -88,9 +88,7 @@ g.test_iproto_insert_arrow_invalid = function(cg)
         --         size: 0}}
         r = _G.iproto_insert_arrow('8210cd020036c70008')
         t.assert_equals(r[box.iproto.key.ERROR_24],
-                        'Failed to decode Arrow IPC data')
-        t.assert_equals(r[box.iproto.key.ERROR][0][1][6].details,
-                        'Unexpected data size')
+                        'Arrow decode error: unexpected data size')
         -- <MP_MAP> {
         --     IPROTO_SPACE_ID: 512,
         --     IPROTO_ARROW: <MP_EXT> {
@@ -99,10 +97,7 @@ g.test_iproto_insert_arrow_invalid = function(cg)
         --         data: [0xde, 0xad, 0xbe, 0xef]}}
         r = _G.iproto_insert_arrow('8210cd020036c70408deadbeef')
         t.assert_equals(r[box.iproto.key.ERROR_24],
-                        'Failed to decode Arrow IPC data')
-        t.assert_equals(r[box.iproto.key.ERROR][0][1][6].method,
-                        'ArrowArrayStreamGetSchema')
-        t.assert_equals(r[box.iproto.key.ERROR][0][1][6].details,
+                        'Arrow decode error: ' ..
                         'Expected at least 8 bytes in remainder of stream')
 
         -- Correct Schema, but Array is missing.
@@ -112,10 +107,7 @@ g.test_iproto_insert_arrow_invalid = function(cg)
             000004000000f0ffffff4000000001000000610000000600080004000c0010000400
             080009000c000c000c0000000400000008000a000c00040006000800ffffffff]])
         t.assert_equals(r[box.iproto.key.ERROR_24],
-                        'Failed to decode Arrow IPC data')
-        t.assert_equals(r[box.iproto.key.ERROR][0][1][6].method,
-                        'ArrowArrayStreamGetNext')
-        t.assert_equals(r[box.iproto.key.ERROR][0][1][6].details,
+                        'Arrow decode error: ' ..
                         'Expected at least 8 bytes in remainder of stream')
 
         -- A valid request, but memtx does not support arrow format.