diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index d4929a372a47b74dbdd5dd8701ac97d3e9429955..9372e5f7f7c22de2150245f1deffe4219a14dc93 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1277,9 +1277,17 @@ vinyl_index_compact(struct index *index)
* If the LSM tree is going to be dropped or truncated on WAL
* recovery, there's no point in replaying statements for it,
* either.
+ *
+ * Note, although we may skip secondary index update in certain
+ * cases (e.g. UPDATE that doesn't touch secondary index parts
+ * or DELETE for which generation of secondary index statement
+ * is deferred), a DML request of any kind always updates the
+ * primary index. Also, we always dump the primary index after
+ * secondary indexes. So we may skip recovery of a DML request
+ * if it has been dumped for the primary index.
*/
static inline bool
-vy_is_committed_one(struct vy_env *env, struct vy_lsm *lsm)
+vy_is_committed(struct vy_env *env, struct vy_lsm *lsm)
{
if (likely(env->status != VINYL_FINAL_RECOVERY_LOCAL))
return false;
@@ -1290,23 +1298,6 @@ vy_is_committed_one(struct vy_env *env, struct vy_lsm *lsm)
return false;
}
-/**
- * Check if a request has already been committed to a space.
- * See also vy_is_committed_one().
- */
-static inline bool
-vy_is_committed(struct vy_env *env, struct space *space)
-{
- if (likely(env->status != VINYL_FINAL_RECOVERY_LOCAL))
- return false;
- for (uint32_t iid = 0; iid < space->index_count; iid++) {
- struct vy_lsm *lsm = vy_lsm(space->index[iid]);
- if (!vy_is_committed_one(env, lsm))
- return false;
- }
- return true;
-}
-
/**
* Get a full tuple by a tuple read from a secondary index.
* @param lsm LSM tree from which the tuple was read.
@@ -1772,11 +1763,11 @@ static int
vy_delete(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
struct space *space, struct request *request)
{
- if (vy_is_committed(env, space))
- return 0;
struct vy_lsm *pk = vy_lsm_find(space, 0);
if (pk == NULL)
return -1;
+ if (vy_is_committed(env, pk))
+ return 0;
struct vy_lsm *lsm = vy_lsm_find_unique(space, request->index_id);
if (lsm == NULL)
return -1;
@@ -1807,7 +1798,7 @@ vy_delete(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
return -1;
for (uint32_t i = 0; i < space->index_count; i++) {
struct vy_lsm *lsm = vy_lsm(space->index[i]);
- if (vy_is_committed_one(env, lsm))
+ if (vy_is_committed(env, lsm))
continue;
rc = vy_tx_set(tx, lsm, delete);
if (rc != 0)
@@ -1891,7 +1882,7 @@ vy_perform_update(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
for (uint32_t i = 1; i < space->index_count; ++i) {
struct vy_lsm *lsm = vy_lsm(space->index[i]);
- if (vy_is_committed_one(env, lsm))
+ if (vy_is_committed(env, lsm))
continue;
if (vy_tx_set_with_colmask(tx, lsm, delete, column_mask) != 0)
goto error;
@@ -1924,7 +1915,10 @@ vy_update(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
struct space *space, struct request *request)
{
assert(tx != NULL && tx->state == VINYL_TX_READY);
- if (vy_is_committed(env, space))
+ struct vy_lsm *pk = vy_lsm_find(space, 0);
+ if (pk == NULL)
+ return -1;
+ if (vy_is_committed(env, pk))
return 0;
struct vy_lsm *lsm = vy_lsm_find_unique(space, request->index_id);
if (lsm == NULL)
@@ -1942,11 +1936,6 @@ vy_update(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
return 0;
/* Apply update operations. */
- struct vy_lsm *pk = vy_lsm(space->index[0]);
- assert(pk != NULL);
- assert(pk->index_id == 0);
- /* Primary key is dumped last. */
- assert(!vy_is_committed_one(env, pk));
uint64_t column_mask = 0;
const char *new_tuple, *new_tuple_end;
uint32_t new_size, old_size;
@@ -2126,7 +2115,10 @@ vy_upsert(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
struct space *space, struct request *request)
{
assert(tx != NULL && tx->state == VINYL_TX_READY);
- if (vy_is_committed(env, space))
+ struct vy_lsm *pk = vy_lsm_find(space, 0);
+ if (pk == NULL)
+ return -1;
+ if (vy_is_committed(env, pk))
return 0;
/* Check update operations. */
if (tuple_update_check_ops(region_aligned_alloc_cb, &fiber()->gc,
@@ -2143,11 +2135,6 @@ vy_upsert(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
const char *tuple_end = request->tuple_end;
const char *ops = request->ops;
const char *ops_end = request->ops_end;
- struct vy_lsm *pk = vy_lsm_find(space, 0);
- if (pk == NULL)
- return -1;
- /* Primary key is dumped last. */
- assert(!vy_is_committed_one(env, pk));
if (tuple_validate_raw(pk->mem_format, tuple))
return -1;
@@ -2240,14 +2227,12 @@ static int
vy_insert(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
struct space *space, struct request *request)
{
- assert(stmt != NULL);
+ assert(tx != NULL && tx->state == VINYL_TX_READY);
struct vy_lsm *pk = vy_lsm_find(space, 0);
if (pk == NULL)
- /* The space hasn't the primary index. */
return -1;
- assert(pk->index_id == 0);
- /* Primary key is dumped last. */
- assert(!vy_is_committed_one(env, pk));
+ if (vy_is_committed(env, pk))
+ return 0;
if (tuple_validate_raw(pk->mem_format, request->tuple))
return -1;
/* First insert into the primary index. */
@@ -2263,7 +2248,7 @@ vy_insert(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
for (uint32_t iid = 1; iid < space->index_count; ++iid) {
struct vy_lsm *lsm = vy_lsm(space->index[iid]);
- if (vy_is_committed_one(env, lsm))
+ if (vy_is_committed(env, lsm))
continue;
if (vy_tx_set(tx, lsm, stmt->new_tuple) != 0)
return -1;
@@ -2290,16 +2275,11 @@ vy_replace(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
struct space *space, struct request *request)
{
assert(tx != NULL && tx->state == VINYL_TX_READY);
- if (vy_is_committed(env, space))
- return 0;
- if (request->type == IPROTO_INSERT)
- return vy_insert(env, tx, stmt, space, request);
-
struct vy_lsm *pk = vy_lsm_find(space, 0);
if (pk == NULL)
return -1;
- /* Primary key is dumped last. */
- assert(!vy_is_committed_one(env, pk));
+ if (vy_is_committed(env, pk))
+ return 0;
/* Validate and create a statement for the new tuple. */
if (tuple_validate_raw(pk->mem_format, request->tuple))
@@ -2352,7 +2332,7 @@ vy_replace(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
}
for (uint32_t i = 1; i < space->index_count; i++) {
struct vy_lsm *lsm = vy_lsm(space->index[i]);
- if (vy_is_committed_one(env, lsm))
+ if (vy_is_committed(env, lsm))
continue;
if (delete != NULL) {
rc = vy_tx_set(tx, lsm, delete);
@@ -2376,7 +2356,12 @@ vinyl_space_execute_replace(struct space *space, struct txn *txn,
struct vy_env *env = vy_env(space->engine);
struct vy_tx *tx = txn->engine_tx;
struct txn_stmt *stmt = txn_current_stmt(txn);
- if (vy_replace(env, tx, stmt, space, request))
+ int rc;
+ if (request->type == IPROTO_INSERT)
+ rc = vy_insert(env, tx, stmt, space, request);
+ else
+ rc = vy_replace(env, tx, stmt, space, request);
+ if (rc != 0)
return -1;
*result = stmt->new_tuple;
return 0;
@@ -3300,6 +3285,8 @@ vinyl_space_apply_initial_join_row(struct space *space, struct request *request)
int rc = -1;
switch (request->type) {
case IPROTO_INSERT:
+ rc = vy_insert(env, tx, &stmt, space, request);
+ break;
case IPROTO_REPLACE:
rc = vy_replace(env, tx, &stmt, space, request);
break;
@@ -4488,7 +4475,7 @@ vy_deferred_delete_on_rollback(struct trigger *trigger, void *event)
* to restore deferred DELETE statements that haven't been dumped
* to disk. To skip deferred DELETEs that have been dumped, we
* use the same technique we employ for normal WAL statements,
- * i.e. we filter them by LSN, see vy_is_committed_one(). To do
+ * i.e. we filter them by LSN, see vy_is_committed(). To do
* that, we need to account the LSN of a WAL row that generated
* a deferred DELETE statement to vy_lsm::dump_lsn, so we install
* an on_commit trigger that propagates the LSN of the WAL row to
@@ -4575,7 +4562,7 @@ vy_deferred_delete_on_replace(struct trigger *trigger, void *event)
struct tuple *region_stmt = NULL;
for (uint32_t i = 1; i < space->index_count; i++) {
struct vy_lsm *lsm = vy_lsm(space->index[i]);
- if (vy_is_committed_one(env, lsm))
+ if (vy_is_committed(env, lsm))
continue;
/*
* As usual, rotate the active in-memory index if
diff --git a/test/vinyl/recover.result b/test/vinyl/recover.result
index a6313f225fa5b523d7c577a5605ae8da72898655..345a019ddc1b017e88eb78ef42675edfdf886b71 100644
--- a/test/vinyl/recover.result
+++ b/test/vinyl/recover.result
@@ -479,3 +479,63 @@ test_run:cmd('cleanup server force_recovery')
---
- true
...
+--
+-- gh-4222: assertion failure while recovering dumped statement
+-- that isn't present in secondary index.
+--
+test_run:cmd('create server test with script = "vinyl/low_quota.lua"')
+---
+- true
+...
+test_run:cmd('start server test with args="1048576"')
+---
+- true
+...
+test_run:cmd('switch test')
+---
+- true
+...
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+pk = s:create_index('primary')
+---
+...
+sk = s:create_index('secondary', {unique = false, parts = {2, 'unsigned'}})
+---
+...
+s:insert{1, 1, 1}
+---
+- [1, 1, 1]
+...
+box.snapshot()
+---
+- ok
+...
+for i = 1, 1024 do s:update(1, {{'=', 3, string.rep('x', 1024)}}) end
+---
+...
+test_run:wait_cond(function() return pk:stat().disk.dump.count == 2 end)
+---
+- true
+...
+sk:stat().disk.dump.count -- 1
+---
+- 1
+...
+test_run:cmd('restart server test with args="1048576"')
+box.space.test:drop()
+---
+...
+test_run:cmd('switch default')
+---
+- true
+...
+test_run:cmd('stop server test')
+---
+- true
+...
+test_run:cmd('cleanup server test')
+---
+- true
+...
diff --git a/test/vinyl/recover.test.lua b/test/vinyl/recover.test.lua
index 1ed266a1af745dcf84f58c9dc0359b1813b6913f..7bdba17826b201ec496301de76ca755325256136 100644
--- a/test/vinyl/recover.test.lua
+++ b/test/vinyl/recover.test.lua
@@ -164,3 +164,30 @@ sum
test_run:cmd('switch default')
test_run:cmd('stop server force_recovery')
test_run:cmd('cleanup server force_recovery')
+
+--
+-- gh-4222: assertion failure while recovering dumped statement
+-- that isn't present in secondary index.
+--
+test_run:cmd('create server test with script = "vinyl/low_quota.lua"')
+test_run:cmd('start server test with args="1048576"')
+test_run:cmd('switch test')
+
+s = box.schema.space.create('test', {engine = 'vinyl'})
+pk = s:create_index('primary')
+sk = s:create_index('secondary', {unique = false, parts = {2, 'unsigned'}})
+
+s:insert{1, 1, 1}
+box.snapshot()
+
+for i = 1, 1024 do s:update(1, {{'=', 3, string.rep('x', 1024)}}) end
+test_run:wait_cond(function() return pk:stat().disk.dump.count == 2 end)
+sk:stat().disk.dump.count -- 1
+
+test_run:cmd('restart server test with args="1048576"')
+
+box.space.test:drop()
+
+test_run:cmd('switch default')
+test_run:cmd('stop server test')
+test_run:cmd('cleanup server test')