[PATCH 5/6] xrow: make NOP requests bodiless

[Vladimir Davydov]

A NOP request has no body, but since it is treated as DML, we still
encode a zero-size map for its body. This complicates conversion of
local requests to NOP in relay as we can't omit xrow_encode_dml (see
the next patch), so let's allow DML requests to be bodiless.

Needed for #3443
---
 src/box/box.cc                   |  1 -
 src/box/lua/xlog.c               | 15 ++++++++-------
 src/box/xrow.c                   | 19 ++++++++++---------
 test/box/before_replace.result   |  4 ++--
 test/box/before_replace.test.lua |  2 +-
 5 files changed, 21 insertions(+), 20 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index 8248cc77..7c8e6bf8 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -273,7 +273,6 @@ recovery_journal_create(struct recovery_journal *journal, struct vclock *v)
 static inline void
 apply_row(struct xstream *stream, struct xrow_header *row)
 {
-	assert(row->bodycnt == 1); /* always 1 for read */
 	(void) stream;
 	struct request request;
 	xrow_decode_dml_xc(row, &request, dml_request_key_map(row->type));
diff --git a/src/box/lua/xlog.c b/src/box/lua/xlog.c
index 70384c1d..2271c829 100644
--- a/src/box/lua/xlog.c
+++ b/src/box/lua/xlog.c
@@ -219,13 +219,14 @@ lbox_xlog_parser_iterate(struct lua_State *L)
 
 	lua_settable(L, -3); /* HEADER */
 
-	assert(row.bodycnt == 1); /* always 1 for read */
-	lua_pushstring(L, "BODY");
-	lua_newtable(L);
-	lbox_xlog_parse_body(L, row.type, (char *)row.body[0].iov_base,
-			     row.body[0].iov_len);
-	lua_settable(L, -3);  /* BODY */
-
+	if (row.bodycnt > 0) {
+		assert(row.bodycnt == 1);
+		lua_pushstring(L, "BODY");
+		lua_newtable(L);
+		lbox_xlog_parse_body(L, row.type, row.body[0].iov_base,
+				     row.body[0].iov_len);
+		lua_settable(L, -3);  /* BODY */
+	}
 	return 2;
 }
 
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 48fbff27..64d845f7 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -405,11 +405,12 @@ int
 xrow_decode_dml(struct xrow_header *row, struct request *request,
 		uint64_t key_map)
 {
-	if (row->bodycnt == 0) {
-		diag_set(ClientError, ER_INVALID_MSGPACK,
-			 "missing request body");
-		return -1;
-	}
+	memset(request, 0, sizeof(*request));
+	request->header = row;
+	request->type = row->type;
+
+	if (row->bodycnt == 0)
+		goto done;
 
 	assert(row->bodycnt == 1);
 	const char *data = (const char *) row->body[0].iov_base;
@@ -422,10 +423,6 @@ error:
 		return -1;
 	}
 
-	memset(request, 0, sizeof(*request));
-	request->header = row;
-	request->type = row->type;
-
 	uint32_t size = mp_decode_map(&data);
 	for (uint32_t i = 0; i < size; i++) {
 		if (! iproto_dml_body_has_key(data, end)) {
@@ -480,6 +477,7 @@ error:
 		diag_set(ClientError, ER_INVALID_MSGPACK, "packet end");
 		return -1;
 	}
+done:
 	if (key_map) {
 		enum iproto_key key = (enum iproto_key) bit_ctz_u64(key_map);
 		diag_set(ClientError, ER_MISSING_REQUEST_FIELD,
@@ -567,6 +565,9 @@ xrow_encode_dml(const struct request *request, struct iovec *iov)
 		map_size++;
 	}
 
+	if (map_size == 0)
+		return 0;
+
 	assert(pos <= begin + len);
 	mp_encode_map(begin, map_size);
 	iov[0].iov_base = begin;
diff --git a/test/box/before_replace.result b/test/box/before_replace.result
index 2b6c1801..cd0a3be1 100644
--- a/test/box/before_replace.result
+++ b/test/box/before_replace.result
@@ -660,9 +660,9 @@ row.HEADER.type
 ---
 - NOP
 ...
-#row.BODY
+row.BODY == nil
 ---
-- 0
+- true
 ...
 -- gh-3128 before_replace with run_triggers
 s2 = box.schema.space.create("test2")
diff --git a/test/box/before_replace.test.lua b/test/box/before_replace.test.lua
index 9b4f49cf..dd464384 100644
--- a/test/box/before_replace.test.lua
+++ b/test/box/before_replace.test.lua
@@ -221,7 +221,7 @@ path = fio.pathjoin(box.cfg.wal_dir, string.format('%020d.xlog', box.info.lsn -
 fun, param, state = xlog.pairs(path)
 state, row = fun(param, state)
 row.HEADER.type
-#row.BODY
+row.BODY == nil
 
 -- gh-3128 before_replace with run_triggers
 s2 = box.schema.space.create("test2")
-- 
2.11.0