[PATCH 6/6] Introduce blackhole engine

[Vladimir Davydov <vdavydov.dev@gmail.com>]

Blackhole is a very simple engine that allows to create spaces that may
written to, but not read from. It only supports INSERT/REPLACE requests.
It doesn't support any indexes hence SELECT is impossible. It does check
space format though and supports on_replace and before_replace triggers.

The whole purpose of this new engine is writing arbitrary rows to WAL
without storing them anywhere. In particular, we need this engine to
write deferred DELETEs generated for vinyl spaces to WAL.

Needed for #2129

@TarantoolBot document
Title: Document blackhole engine
Spaces created with blackhole engine may be written to, but not read
from (they don't even support space.create_index command). Operations
done on a blackhole space are written to WAL and hence get replicated,
but not stored anywhere else.
---
 src/box/CMakeLists.txt      |   1 +
 src/box/blackhole.c         | 210 ++++++++++++++++++++++++++++++++++++++++++++
 src/box/blackhole.h         |  58 ++++++++++++
 src/box/box.cc              |   4 +
 test/box/blackhole.result   | 177 +++++++++++++++++++++++++++++++++++++
 test/box/blackhole.test.lua |  64 ++++++++++++++
 6 files changed, 514 insertions(+)
 create mode 100644 src/box/blackhole.c
 create mode 100644 src/box/blackhole.h
 create mode 100644 test/box/blackhole.result
 create mode 100644 test/box/blackhole.test.lua

diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
index fabeb3fd..ad544270 100644
--- a/src/box/CMakeLists.txt
+++ b/src/box/CMakeLists.txt
@@ -69,6 +69,7 @@ add_library(box STATIC
     memtx_engine.c
     memtx_space.c
     sysview.c
+    blackhole.c
     vinyl.c
     vy_stmt.c
     vy_mem.c
diff --git a/src/box/blackhole.c b/src/box/blackhole.c
new file mode 100644
index 00000000..96b9fd14
--- /dev/null
+++ b/src/box/blackhole.c
@@ -0,0 +1,210 @@
+/*
+ * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "blackhole.h"
+
+#include <stddef.h>
+#include <stdlib.h>
+#include <small/rlist.h>
+
+#include "diag.h"
+#include "error.h"
+#include "errcode.h"
+#include "engine.h"
+#include "space.h"
+#include "txn.h"
+#include "tuple.h"
+#include "xrow.h"
+
+static void
+blackhole_space_destroy(struct space *space)
+{
+	free(space);
+}
+
+static int
+blackhole_space_execute_replace(struct space *space, struct txn *txn,
+				struct request *request, struct tuple **result)
+{
+	struct txn_stmt *stmt = txn_current_stmt(txn);
+	stmt->new_tuple = tuple_new(space->format, request->tuple,
+				    request->tuple_end);
+	if (stmt->new_tuple == NULL)
+		return -1;
+	tuple_ref(stmt->new_tuple);
+	*result = stmt->new_tuple;
+	return 0;
+}
+
+static int
+blackhole_space_execute_delete(struct space *space, struct txn *txn,
+			       struct request *request, struct tuple **result)
+{
+	(void)space;
+	(void)txn;
+	(void)request;
+	(void)result;
+	diag_set(ClientError, ER_UNSUPPORTED, "Blackhole", "delete()");
+	return -1;
+}
+
+static int
+blackhole_space_execute_update(struct space *space, struct txn *txn,
+			       struct request *request, struct tuple **result)
+{
+	(void)space;
+	(void)txn;
+	(void)request;
+	(void)result;
+	diag_set(ClientError, ER_UNSUPPORTED, "Blackhole", "update()");
+	return -1;
+}
+
+static int
+blackhole_space_execute_upsert(struct space *space, struct txn *txn,
+			       struct request *request)
+{
+	(void)space;
+	(void)txn;
+	(void)request;
+	diag_set(ClientError, ER_UNSUPPORTED, "Blackhole", "upsert()");
+	return -1;
+}
+
+static struct index *
+blackhole_space_create_index(struct space *space, struct index_def *def)
+{
+	(void)space;
+	(void)def;
+	/* See blackhole_engine_create_space(). */
+	unreachable();
+	return NULL;
+}
+
+static const struct space_vtab blackhole_space_vtab = {
+	/* .destroy = */ blackhole_space_destroy,
+	/* .bsize = */ generic_space_bsize,
+	/* .apply_initial_join_row = */ generic_space_apply_initial_join_row,
+	/* .execute_replace = */ blackhole_space_execute_replace,
+	/* .execute_delete = */ blackhole_space_execute_delete,
+	/* .execute_update = */ blackhole_space_execute_update,
+	/* .execute_upsert = */ blackhole_space_execute_upsert,
+	/* .init_system_space = */ generic_init_system_space,
+	/* .check_index_def = */ generic_space_check_index_def,
+	/* .create_index = */ blackhole_space_create_index,
+	/* .add_primary_key = */ generic_space_add_primary_key,
+	/* .drop_primary_key = */ generic_space_drop_primary_key,
+	/* .check_format = */ generic_space_check_format,
+	/* .build_index = */ generic_space_build_index,
+	/* .swap_index = */ generic_space_swap_index,
+	/* .prepare_alter = */ generic_space_prepare_alter,
+};
+
+static void
+blackhole_engine_shutdown(struct engine *engine)
+{
+	free(engine);
+}
+
+static struct space *
+blackhole_engine_create_space(struct engine *engine, struct space_def *def,
+			      struct rlist *key_list)
+{
+	if (!rlist_empty(key_list)) {
+		diag_set(ClientError, ER_UNSUPPORTED, "Blackhole", "indexes");
+		return NULL;
+	}
+
+	struct space *space = (struct space *)calloc(1, sizeof(*space));
+	if (space == NULL) {
+		diag_set(OutOfMemory, sizeof(*space),
+			 "malloc", "struct space");
+		return NULL;
+	}
+
+	/* Allocate tuples on runtime arena, but check space format. */
+	struct tuple_format *format;
+	format = tuple_format_new(&tuple_format_runtime->vtab, NULL, 0, 0,
+				  def->fields, def->field_count, def->dict);
+	if (format == NULL) {
+		free(space);
+		return NULL;
+	}
+	format->exact_field_count = def->exact_field_count;
+	tuple_format_ref(format);
+
+	if (space_create(space, engine, &blackhole_space_vtab,
+			 def, key_list, format) != 0) {
+		tuple_format_unref(format);
+		free(space);
+		return NULL;
+	}
+	return space;
+}
+
+static const struct engine_vtab blackhole_engine_vtab = {
+	/* .shutdown = */ blackhole_engine_shutdown,
+	/* .create_space = */ blackhole_engine_create_space,
+	/* .join = */ generic_engine_join,
+	/* .begin = */ generic_engine_begin,
+	/* .begin_statement = */ generic_engine_begin_statement,
+	/* .prepare = */ generic_engine_prepare,
+	/* .commit = */ generic_engine_commit,
+	/* .rollback_statement = */ generic_engine_rollback_statement,
+	/* .rollback = */ generic_engine_rollback,
+	/* .bootstrap = */ generic_engine_bootstrap,
+	/* .begin_initial_recovery = */ generic_engine_begin_initial_recovery,
+	/* .begin_final_recovery = */ generic_engine_begin_final_recovery,
+	/* .end_recovery = */ generic_engine_end_recovery,
+	/* .begin_checkpoint = */ generic_engine_begin_checkpoint,
+	/* .wait_checkpoint = */ generic_engine_wait_checkpoint,
+	/* .commit_checkpoint = */ generic_engine_commit_checkpoint,
+	/* .abort_checkpoint = */ generic_engine_abort_checkpoint,
+	/* .collect_garbage = */ generic_engine_collect_garbage,
+	/* .backup = */ generic_engine_backup,
+	/* .memory_stat = */ generic_engine_memory_stat,
+	/* .reset_stat = */ generic_engine_reset_stat,
+	/* .check_space_def = */ generic_engine_check_space_def,
+};
+
+struct engine *
+blackhole_engine_new(void)
+{
+	struct engine *engine = calloc(1, sizeof(*engine));
+	if (engine == NULL) {
+		diag_set(OutOfMemory, sizeof(*engine),
+			 "malloc", "struct engine");
+		return NULL;
+	}
+
+	engine->vtab = &blackhole_engine_vtab;
+	engine->name = "blackhole";
+	return engine;
+}
diff --git a/src/box/blackhole.h b/src/box/blackhole.h
new file mode 100644
index 00000000..5a78610d
--- /dev/null
+++ b/src/box/blackhole.h
@@ -0,0 +1,58 @@
+#ifndef TARANTOOL_BOX_BLACKHOLE_H_INCLUDED
+#define TARANTOOL_BOX_BLACKHOLE_H_INCLUDED
+/*
+ * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include <stddef.h>
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+struct engine *
+blackhole_engine_new(void);
+
+#if defined(__cplusplus)
+} /* extern "C" */
+
+#include "diag.h"
+
+static inline struct engine *
+blackhole_engine_new_xc(void)
+{
+	struct engine *engine = blackhole_engine_new();
+	if (engine == NULL)
+		diag_raise();
+	return engine;
+}
+
+#endif /* defined(__plusplus) */
+
+#endif /* TARANTOOL_BOX_BLACKHOLE_H_INCLUDED */
diff --git a/src/box/box.cc b/src/box/box.cc
index 9c3f87f0..22dc64d4 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -51,6 +51,7 @@
 #include "engine.h"
 #include "memtx_engine.h"
 #include "sysview.h"
+#include "blackhole.h"
 #include "vinyl.h"
 #include "space.h"
 #include "index.h"
@@ -1627,6 +1628,9 @@ engine_init()
 	struct sysview_engine *sysview = sysview_engine_new_xc();
 	engine_register((struct engine *)sysview);
 
+	struct engine *blackhole = blackhole_engine_new_xc();
+	engine_register(blackhole);
+
 	struct vinyl_engine *vinyl;
 	vinyl = vinyl_engine_new_xc(cfg_gets("vinyl_dir"),
 				    cfg_geti64("vinyl_memory"),
diff --git a/test/box/blackhole.result b/test/box/blackhole.result
new file mode 100644
index 00000000..a985fcd7
--- /dev/null
+++ b/test/box/blackhole.result
@@ -0,0 +1,177 @@
+test_run = require('test_run').new()
+---
+...
+s = box.schema.space.create('test', {engine = 'blackhole'})
+---
+...
+-- Blackhole doesn't support indexes.
+s:create_index('pk')
+---
+- error: Blackhole does not support indexes
+...
+-- Blackhole does support space format.
+s:format{{'key', 'unsigned'}, {'value', 'string'}}
+---
+...
+s:format()
+---
+- [{'name': 'key', 'type': 'unsigned'}, {'name': 'value', 'type': 'string'}]
+...
+t = s:insert{1, 'a'} -- ok
+---
+...
+t, t.key, t.value
+---
+- [1, 'a']
+- 1
+- a
+...
+s:insert{1, 2, 3} -- error
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected string'
+...
+s:replace{'a', 'b', 'c'} -- error
+---
+- error: 'Tuple field 1 type does not match one required by operation: expected unsigned'
+...
+s:format{}
+---
+...
+s:insert{1, 2, 3} -- ok
+---
+- [1, 2, 3]
+...
+s:replace{'a', 'b', 'c'} -- ok
+---
+- ['a', 'b', 'c']
+...
+-- Blackhole doesn't support delete/update/upsert operations.
+box.internal.delete(s.id, 0, {})
+---
+- error: Blackhole does not support delete()
+...
+box.internal.update(s.id, 0, {}, {})
+---
+- error: Blackhole does not support update()
+...
+box.internal.upsert(s.id, {}, {})
+---
+- error: Blackhole does not support upsert()
+...
+-- Blackhole supports on_replace and before_replace triggers.
+s_old = nil
+---
+...
+s_new = nil
+---
+...
+f1 = s:on_replace(function(old, new) s_old = old s_new = new end)
+---
+...
+s:replace{1, 2, 3}
+---
+- [1, 2, 3]
+...
+s_old, s_new
+---
+- null
+- [1, 2, 3]
+...
+f2 = s:before_replace(function(old, new) return box.tuple.new{4, 5, 6} end)
+---
+...
+s:replace{1, 2, 3}
+---
+- [4, 5, 6]
+...
+s_old, s_new
+---
+- null
+- [4, 5, 6]
+...
+s:on_replace(nil, f1)
+---
+...
+s:before_replace(nil, f2)
+---
+...
+-- Test recovery.
+test_run:cmd('restart server default')
+s = box.space.test
+---
+...
+-- Test snapshot.
+box.snapshot()
+---
+- ok
+...
+-- Operations done on a blackhole space are written to the WAL
+-- and therefore get replicated. Check it with the aid of an
+-- on_replace trigger.
+box.schema.user.grant('guest', 'replication')
+---
+...
+test_run:cmd("create server replica with rpl_master=default, script='replication/replica.lua'")
+---
+- true
+...
+test_run:cmd("start server replica")
+---
+- true
+...
+test_run:cmd("switch replica")
+---
+- true
+...
+t = {}
+---
+...
+_ = box.space.test:on_replace(function(old, new) table.insert(t, new) end)
+---
+...
+test_run:cmd('switch default')
+---
+- true
+...
+s = box.space.test
+---
+...
+for i = 1, 5 do s:replace{i} end
+---
+...
+vclock = test_run:get_vclock('default')
+---
+...
+test_run:wait_vclock('replica', vclock)
+---
+...
+test_run:cmd("switch replica")
+---
+- true
+...
+t
+---
+- - [1]
+  - [2]
+  - [3]
+  - [4]
+  - [5]
+...
+test_run:cmd('switch default')
+---
+- true
+...
+test_run:cmd("stop server replica")
+---
+- true
+...
+test_run:cmd("cleanup server replica")
+---
+- true
+...
+box.schema.user.revoke('guest', 'replication')
+---
+...
+s:drop()
+---
+...
diff --git a/test/box/blackhole.test.lua b/test/box/blackhole.test.lua
new file mode 100644
index 00000000..5dcf9e61
--- /dev/null
+++ b/test/box/blackhole.test.lua
@@ -0,0 +1,64 @@
+test_run = require('test_run').new()
+
+s = box.schema.space.create('test', {engine = 'blackhole'})
+
+-- Blackhole doesn't support indexes.
+s:create_index('pk')
+
+-- Blackhole does support space format.
+s:format{{'key', 'unsigned'}, {'value', 'string'}}
+s:format()
+t = s:insert{1, 'a'} -- ok
+t, t.key, t.value
+s:insert{1, 2, 3} -- error
+s:replace{'a', 'b', 'c'} -- error
+s:format{}
+s:insert{1, 2, 3} -- ok
+s:replace{'a', 'b', 'c'} -- ok
+
+-- Blackhole doesn't support delete/update/upsert operations.
+box.internal.delete(s.id, 0, {})
+box.internal.update(s.id, 0, {}, {})
+box.internal.upsert(s.id, {}, {})
+
+-- Blackhole supports on_replace and before_replace triggers.
+s_old = nil
+s_new = nil
+f1 = s:on_replace(function(old, new) s_old = old s_new = new end)
+s:replace{1, 2, 3}
+s_old, s_new
+f2 = s:before_replace(function(old, new) return box.tuple.new{4, 5, 6} end)
+s:replace{1, 2, 3}
+s_old, s_new
+s:on_replace(nil, f1)
+s:before_replace(nil, f2)
+
+-- Test recovery.
+test_run:cmd('restart server default')
+s = box.space.test
+
+-- Test snapshot.
+box.snapshot()
+
+-- Operations done on a blackhole space are written to the WAL
+-- and therefore get replicated. Check it with the aid of an
+-- on_replace trigger.
+box.schema.user.grant('guest', 'replication')
+test_run:cmd("create server replica with rpl_master=default, script='replication/replica.lua'")
+test_run:cmd("start server replica")
+test_run:cmd("switch replica")
+t = {}
+_ = box.space.test:on_replace(function(old, new) table.insert(t, new) end)
+test_run:cmd('switch default')
+s = box.space.test
+for i = 1, 5 do s:replace{i} end
+vclock = test_run:get_vclock('default')
+test_run:wait_vclock('replica', vclock)
+test_run:cmd("switch replica")
+t
+test_run:cmd('switch default')
+test_run:cmd("stop server replica")
+test_run:cmd("cleanup server replica")
+box.schema.user.revoke('guest', 'replication')
+
+s:drop()
-- 
2.11.0