10992975672

Committed 23 Sep 2024 11:03AM UTC coverage: 85.929% (+0.02%) from 85.906%

Build # 10992975672

Build Type

push

github

Committed by

locker

Commit Message

box: check fiber slice in generic implementation of index count

`index.count()` may hang for too long in Vinyl if a substantial
consecutive hunk of the space is stored in memory. Let's add
a fiber slice check to it to prevent it from blocking the TX thread
for too long.

Closes #10553

NO_DOC=bug fix

(cherry picked from commit e19bca5a7)

Run Details

63066 of 114355 branches covered (55.15%)

4 of 4 new or added lines in 1 file covered. (100.0%)

22 existing lines in 7 files now uncovered.

93935 of 109317 relevant lines covered (85.93%)

2622314.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.96

/src/box/vy_point_lookup.c

/*
 * Copyright 2010-2017, 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 AUTHORS ``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
 * AUTHORS 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 "vy_point_lookup.h"

#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>

#include <small/region.h>
#include <small/rlist.h>

#include "fiber.h"

#include "vy_lsm.h"
#include "vy_stmt.h"
#include "vy_tx.h"
#include "vy_mem.h"
#include "vy_run.h"
#include "vy_cache.h"
#include "vy_history.h"

/**
 * Scan TX write set for given key.
 * Add one or no statement to the history list.
 */
static int
vy_point_lookup_scan_txw(struct vy_lsm *lsm, struct vy_tx *tx,
                         struct vy_entry key, struct vy_history *history)
{
        if (tx == NULL)
                return 0;
        lsm->stat.txw.iterator.lookup++;
        struct txv *txv =
                write_set_search_key(&tx->write_set, lsm, key);
        assert(txv == NULL || txv->lsm == lsm);
        if (txv == NULL)
                return 0;
        vy_stmt_counter_acct_tuple(&lsm->stat.txw.iterator.get,
                                   txv->entry.stmt);
        return vy_history_append_stmt(history, txv->entry);
}

/**
 * Scan LSM tree cache for given key.
 * Add one or no statement to the history list.
 */
static int
vy_point_lookup_scan_cache(struct vy_lsm *lsm, const struct vy_read_view **rv,
                           bool is_prepared_ok, struct vy_entry key,
                           struct vy_history *history)
{
        lsm->cache.stat.lookup++;
        struct vy_entry entry = vy_cache_get(&lsm->cache, key);

        if (entry.stmt == NULL || vy_stmt_lsn(entry.stmt) > (*rv)->vlsn ||
            (!is_prepared_ok && vy_stmt_is_prepared(entry.stmt)))
                return 0;

        vy_stmt_counter_acct_tuple(&lsm->cache.stat.get, entry.stmt);
        return vy_history_append_stmt(history, entry);
}

/**
 * Scan one particular mem.
 * Add found statements to the history list up to terminal statement.
 */
static int
vy_point_lookup_scan_mem(struct vy_lsm *lsm, struct vy_mem *mem,
                         const struct vy_read_view **rv, bool is_prepared_ok,
                         struct vy_entry key, struct vy_history *history,
                         int64_t *min_skipped_plsn)
{
        struct vy_mem_iterator mem_itr;
        vy_mem_iterator_open(&mem_itr, &lsm->stat.memory.iterator,
                             mem, ITER_EQ, key, rv, is_prepared_ok);
        struct vy_history mem_history;
        vy_history_create(&mem_history, &lsm->env->history_node_pool);
        int rc = vy_mem_iterator_next(&mem_itr, &mem_history);
        vy_history_splice(history, &mem_history);
        *min_skipped_plsn = MIN(*min_skipped_plsn, mem_itr.min_skipped_plsn);
        vy_mem_iterator_close(&mem_itr);
        return rc;

}

/**
 * Scan all mems that belongs to the LSM tree.
 * Add found statements to the history list up to terminal statement.
 */
static int
vy_point_lookup_scan_mems(struct vy_lsm *lsm, struct vy_tx *tx,
                          const struct vy_read_view **rv, bool is_prepared_ok,
                          struct vy_entry key, struct vy_history *history)
{
        assert(lsm->mem != NULL);
        int64_t min_skipped_plsn = INT64_MAX;
        if (vy_point_lookup_scan_mem(lsm, lsm->mem, rv, is_prepared_ok,
                                     key, history, &min_skipped_plsn) != 0)
                return -1;
        struct vy_mem *mem;
        rlist_foreach_entry(mem, &lsm->sealed, in_sealed) {
                if (vy_history_is_terminal(history))
                        break;
                if (vy_point_lookup_scan_mem(lsm, mem, rv, is_prepared_ok,
                                             key, history,
                                             &min_skipped_plsn) != 0)
                        return -1;
        }
        /*
         * Switch to read view if we skipped a prepared statement.
         */
        if (tx != NULL && min_skipped_plsn != INT64_MAX) {
                if (vy_tx_send_to_read_view(tx, min_skipped_plsn) != 0)
                        return -1;
                if (tx->state == VINYL_TX_ABORT) {
                        diag_set(ClientError, ER_TRANSACTION_CONFLICT);
                        return -1;
                }
        }
        return 0;
}

/**
 * Scan one particular slice.
 * Add found statements to the history list up to terminal statement.
 */
static int
vy_point_lookup_scan_slice(struct vy_lsm *lsm, struct vy_slice *slice,
                           const struct vy_read_view **rv, struct vy_entry key,
                           struct vy_history *history)
{
        /*
         * The format of the statement must be exactly the space
         * format with the same identifier to fully match the
         * format in vy_mem.
         */
        struct vy_run_iterator run_itr;
        vy_run_iterator_open(&run_itr, &lsm->stat.disk.iterator, slice,
                             ITER_EQ, key, rv, lsm->cmp_def, lsm->key_def,
                             lsm->disk_format);
        struct vy_history slice_history;
        vy_history_create(&slice_history, &lsm->env->history_node_pool);
        int rc = vy_run_iterator_next(&run_itr, &slice_history);
        vy_history_splice(history, &slice_history);
        vy_run_iterator_close(&run_itr);
        return rc;
}

/**
 * Find a range and scan all slices that belongs to the range.
 * Add found statements to the history list up to terminal statement.
 * All slices are pinned before first slice scan, so it's guaranteed
 * that complete history from runs will be extracted.
 */
static int
vy_point_lookup_scan_slices(struct vy_lsm *lsm, const struct vy_read_view **rv,
                            struct vy_entry key, struct vy_history *history)
{
        struct vy_range *range = vy_range_tree_find_by_key(&lsm->range_tree,
                                                           ITER_EQ, key);
        assert(range != NULL);
        int slice_count = range->slice_count;
        size_t region_svp = region_used(&fiber()->gc);
        if (slice_count == 0)
                return 0;
        struct vy_slice **slices =
                xregion_alloc_array(&fiber()->gc, typeof(slices[0]),
                                    slice_count);
        int i = 0;
        struct vy_slice *slice;
        rlist_foreach_entry(slice, &range->slices, in_range) {
                vy_slice_pin(slice);
                slices[i++] = slice;
        }
        assert(i == slice_count);
        ERROR_INJECT_YIELD(ERRINJ_VY_POINT_LOOKUP_DELAY);
        int rc = 0;
        for (i = 0; i < slice_count; i++) {
                if (rc == 0 && !vy_history_is_terminal(history))
                        rc = vy_point_lookup_scan_slice(lsm, slices[i],
                                                        rv, key, history);
                vy_slice_unpin(slices[i]);
        }
        region_truncate(&fiber()->gc, region_svp);
        return rc;
}

int
vy_point_lookup(struct vy_lsm *lsm, struct vy_tx *tx,
                const struct vy_read_view **rv,
                struct vy_entry key, struct vy_entry *ret)
{
        /* All key parts must be set for a point lookup. */
        assert(vy_stmt_is_full_key(key.stmt, lsm->cmp_def));
        assert(tx == NULL || tx->state == VINYL_TX_READY);

        *ret = vy_entry_none();
        int rc = 0;

        /* History list */
        struct vy_history history, mem_history, disk_history;
        vy_history_create(&history, &lsm->env->history_node_pool);
        vy_history_create(&mem_history, &lsm->env->history_node_pool);
        vy_history_create(&disk_history, &lsm->env->history_node_pool);

        rc = vy_point_lookup_scan_txw(lsm, tx, key, &history);
        if (rc != 0 || vy_history_is_terminal(&history))
                goto done;

        bool is_prepared_ok = tx != NULL ? vy_tx_is_prepared_ok(tx) : false;
        rc = vy_point_lookup_scan_cache(lsm, rv, is_prepared_ok, key, &history);
        if (rc != 0 || vy_history_is_terminal(&history))
                goto done;

restart:
        rc = vy_point_lookup_scan_mems(lsm, tx, rv, is_prepared_ok,
                                       key, &mem_history);
        if (rc != 0 || vy_history_is_terminal(&mem_history))
                goto done;

        /* Save version before yield */
        uint32_t mem_version = lsm->mem->version;
        uint32_t mem_list_version = lsm->mem_list_version;

        rc = vy_point_lookup_scan_slices(lsm, rv, key, &disk_history);
        if (rc != 0)
                goto done;

        if (tx != NULL && tx->state == VINYL_TX_ABORT) {
                /*
                 * The transaction was aborted while we were reading
                 * disk. We must stop now and return an error as this
                 * function could be called by a DML request aborted
                 * by a DDL operation: failing early will prevent it
                 * from dereferencing a destroyed space.
                 */
                diag_set(ClientError, ER_TRANSACTION_CONFLICT);
                rc = -1;
                goto done;
        }

        if (mem_list_version != lsm->mem_list_version) {
                /*
                 * Mem list was changed during yield. This could be rotation
                 * or a dump. In case of dump the memory referenced by
                 * statement history is gone and we need to reread new history.
                 * This in unnecessary in case of rotation but since we
                 * cannot distinguish these two cases we always restart.
                 */
                vy_history_cleanup(&mem_history);
                vy_history_cleanup(&disk_history);
                goto restart;
        }

        if (mem_version != lsm->mem->version) {
                /*
                 * Rescan the memory level if its version changed while we
                 * were reading disk, because there may be new statements
                 * matching the search key.
                 */
                vy_history_cleanup(&mem_history);
                rc = vy_point_lookup_scan_mems(lsm, tx, rv, is_prepared_ok,
                                               key, &mem_history);
                if (rc != 0)
                        goto done;
                if (vy_history_is_terminal(&mem_history))
                        vy_history_cleanup(&disk_history);
        }

done:
        vy_history_splice(&history, &mem_history);
        vy_history_splice(&history, &disk_history);

        if (rc == 0) {
                int upserts_applied;
                rc = vy_history_apply(&history, lsm->cmp_def,
                                      false, &upserts_applied, ret);
                lsm->stat.upsert.applied += upserts_applied;
        }
        vy_history_cleanup(&history);

        if (rc != 0)
                return -1;

        return 0;
}

int
vy_point_lookup_mem(struct vy_lsm *lsm, const struct vy_read_view **rv,
                    struct vy_entry key, struct vy_entry *ret)
{
        assert(vy_stmt_is_full_key(key.stmt, lsm->cmp_def));

        int rc;
        struct vy_history history;
        vy_history_create(&history, &lsm->env->history_node_pool);

        rc = vy_point_lookup_scan_cache(lsm, rv, /*is_prepared_ok=*/true,
                                        key, &history);
        if (rc != 0 || vy_history_is_terminal(&history))
                goto done;

        rc = vy_point_lookup_scan_mems(lsm, /*tx=*/NULL, rv,
                                       /*is_prepared_ok=*/true, key, &history);
        if (rc != 0 || vy_history_is_terminal(&history))
                goto done;

        *ret = vy_entry_none();
        goto out;
done:
        if (rc == 0) {
                int upserts_applied;
                rc = vy_history_apply(&history, lsm->cmp_def,
                                      true, &upserts_applied, ret);
                lsm->stat.upsert.applied += upserts_applied;
        }
out:
        vy_history_cleanup(&history);
        return rc;
}

1	/*
2	* Copyright 2010-2017, Tarantool AUTHORS, please see AUTHORS file.
3	*
4	* Redistribution and use in source and binary forms, with or
5	* without modification, are permitted provided that the following
6	* conditions are met:
7	*
8	* 1. Redistributions of source code must retain the above
9	* copyright notice, this list of conditions and the
10	* following disclaimer.
11	*
12	* 2. Redistributions in binary form must reproduce the above
13	* copyright notice, this list of conditions and the following
14	* disclaimer in the documentation and/or other materials
15	* provided with the distribution.
16	*
17	* THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
18	* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
19	* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
20	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
21	* AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
22	* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23	* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
25	* BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
26	* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
28	* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
29	* SUCH DAMAGE.
30	*/
31	#include "vy_point_lookup.h"
32
33	#include <stdbool.h>
34	#include <stddef.h>
35	#include <stdint.h>
36
37	#include <small/region.h>
38	#include <small/rlist.h>
39
40	#include "fiber.h"
41
42	#include "vy_lsm.h"
43	#include "vy_stmt.h"
44	#include "vy_tx.h"
45	#include "vy_mem.h"
46	#include "vy_run.h"
47	#include "vy_cache.h"
48	#include "vy_history.h"
49
50	/**
51	* Scan TX write set for given key.
52	* Add one or no statement to the history list.
53	*/
54	static int
55	vy_point_lookup_scan_txw(struct vy_lsm lsm, struct vy_tx tx,	1,590,900✔
56	struct vy_entry key, struct vy_history *history)
57	{
58	if (tx == NULL)	1,590,900✔
59	return 0;	97✔
60	lsm->stat.txw.iterator.lookup++;	1,590,800✔
61	struct txv *txv =
62	write_set_search_key(&tx->write_set, lsm, key);	1,590,800✔
63	assert(txv == NULL \|\| txv->lsm == lsm);	1,590,800!
64	if (txv == NULL)	1,590,800✔
65	return 0;	1,585,380✔
66	vy_stmt_counter_acct_tuple(&lsm->stat.txw.iterator.get,	5,423✔
67	txv->entry.stmt);
68	return vy_history_append_stmt(history, txv->entry);	5,423✔
69	}
70
71	/**
72	* Scan LSM tree cache for given key.
73	* Add one or no statement to the history list.
74	*/
75	static int
76	vy_point_lookup_scan_cache(struct vy_lsm lsm, const struct vy_read_view *rv,	1,618,240✔
77	bool is_prepared_ok, struct vy_entry key,
78	struct vy_history *history)
79	{
80	lsm->cache.stat.lookup++;	1,618,240✔
81	struct vy_entry entry = vy_cache_get(&lsm->cache, key);	1,618,240!
82
83	if (entry.stmt == NULL \|\| vy_stmt_lsn(entry.stmt) > (*rv)->vlsn \|\|	1,618,240!
84	(!is_prepared_ok && vy_stmt_is_prepared(entry.stmt)))	518,561!
85	return 0;	1,618,240✔
86
87	vy_stmt_counter_acct_tuple(&lsm->cache.stat.get, entry.stmt);	518,561!
88	return vy_history_append_stmt(history, entry);	518,561!
89	}
90
91	/**
92	* Scan one particular mem.
93	* Add found statements to the history list up to terminal statement.
94	*/
95	static int
96	vy_point_lookup_scan_mem(struct vy_lsm lsm, struct vy_mem mem,	1,678,340✔
97	const struct vy_read_view **rv, bool is_prepared_ok,
98	struct vy_entry key, struct vy_history *history,
99	int64_t *min_skipped_plsn)
100	{
101	struct vy_mem_iterator mem_itr;
102	vy_mem_iterator_open(&mem_itr, &lsm->stat.memory.iterator,	1,678,340!
103	mem, ITER_EQ, key, rv, is_prepared_ok);
104	struct vy_history mem_history;
105	vy_history_create(&mem_history, &lsm->env->history_node_pool);	1,678,340!
106	int rc = vy_mem_iterator_next(&mem_itr, &mem_history);	1,678,340!
107	vy_history_splice(history, &mem_history);	1,678,340!
108	min_skipped_plsn = MIN(min_skipped_plsn, mem_itr.min_skipped_plsn);	1,678,340✔
109	vy_mem_iterator_close(&mem_itr);	1,678,340!
110	return rc;	1,678,340✔
111
112	}
113
114	/**
115	* Scan all mems that belongs to the LSM tree.
116	* Add found statements to the history list up to terminal statement.
117	*/
118	static int
119	vy_point_lookup_scan_mems(struct vy_lsm lsm, struct vy_tx tx,	1,123,700✔
120	const struct vy_read_view **rv, bool is_prepared_ok,
121	struct vy_entry key, struct vy_history *history)
122	{
123	assert(lsm->mem != NULL);	1,123,700!
124	int64_t min_skipped_plsn = INT64_MAX;	1,123,700✔
125	if (vy_point_lookup_scan_mem(lsm, lsm->mem, rv, is_prepared_ok,	1,123,700!
126	key, history, &min_skipped_plsn) != 0)
127	return -1;	1,123,700✔
128	struct vy_mem *mem;
129	rlist_foreach_entry(mem, &lsm->sealed, in_sealed) {	3,356,680✔
130	if (vy_history_is_terminal(history))	720,869!
131	break;	166,227✔
132	if (vy_point_lookup_scan_mem(lsm, mem, rv, is_prepared_ok,	554,642!
133	key, history,
134	&min_skipped_plsn) != 0)
135	return -1;	×
136	}
137	/*
138	* Switch to read view if we skipped a prepared statement.
139	*/
140	if (tx != NULL && min_skipped_plsn != INT64_MAX) {	1,123,700✔
141	if (vy_tx_send_to_read_view(tx, min_skipped_plsn) != 0)	1!
142	return -1;	×
143	if (tx->state == VINYL_TX_ABORT) {	1!
UNCOV 144	diag_set(ClientError, ER_TRANSACTION_CONFLICT);	×
UNCOV 145	return -1;	×
146	}
147	}
148	return 0;	1,123,700✔
149	}
150
151	/**
152	* Scan one particular slice.
153	* Add found statements to the history list up to terminal statement.
154	*/
155	static int
156	vy_point_lookup_scan_slice(struct vy_lsm lsm, struct vy_slice slice,	407,547✔
157	const struct vy_read_view **rv, struct vy_entry key,
158	struct vy_history *history)
159	{
160	/*
161	* The format of the statement must be exactly the space
162	* format with the same identifier to fully match the
163	* format in vy_mem.
164	*/
165	struct vy_run_iterator run_itr;
166	vy_run_iterator_open(&run_itr, &lsm->stat.disk.iterator, slice,	407,547!
167	ITER_EQ, key, rv, lsm->cmp_def, lsm->key_def,
168	lsm->disk_format);
169	struct vy_history slice_history;
170	vy_history_create(&slice_history, &lsm->env->history_node_pool);	407,547!
171	int rc = vy_run_iterator_next(&run_itr, &slice_history);	407,547!
172	vy_history_splice(history, &slice_history);	407,546!
173	vy_run_iterator_close(&run_itr);	407,546!
174	return rc;	407,546✔
175	}
176
177	/**
178	* Find a range and scan all slices that belongs to the range.
179	* Add found statements to the history list up to terminal statement.
180	* All slices are pinned before first slice scan, so it's guaranteed
181	* that complete history from runs will be extracted.
182	*/
183	static int
184	vy_point_lookup_scan_slices(struct vy_lsm lsm, const struct vy_read_view *rv,	895,084✔
185	struct vy_entry key, struct vy_history *history)
186	{
187	struct vy_range *range = vy_range_tree_find_by_key(&lsm->range_tree,	895,084✔
188	ITER_EQ, key);
189	assert(range != NULL);	895,084!
190	int slice_count = range->slice_count;	895,084✔
191	size_t region_svp = region_used(&fiber()->gc);	895,084!
192	if (slice_count == 0)	895,084✔
193	return 0;	719,763✔
194	struct vy_slice **slices =	175,321✔
195	xregion_alloc_array(&fiber()->gc, typeof(slices[0]),	175,321!
196	slice_count);
197	int i = 0;	175,321✔
198	struct vy_slice *slice;
199	rlist_foreach_entry(slice, &range->slices, in_range) {	773,738✔
200	vy_slice_pin(slice);	423,096✔
201	slices[i++] = slice;	423,096✔
202	}
203	assert(i == slice_count);	175,321!
204	ERROR_INJECT_YIELD(ERRINJ_VY_POINT_LOOKUP_DELAY);	175,323!
205	int rc = 0;	175,321✔
206	for (i = 0; i < slice_count; i++) {	598,416✔
207	if (rc == 0 && !vy_history_is_terminal(history))	423,096!
208	rc = vy_point_lookup_scan_slice(lsm, slices[i],	407,547✔
209	rv, key, history);
210	vy_slice_unpin(slices[i]);	423,095✔
211	}
212	region_truncate(&fiber()->gc, region_svp);	175,320!
213	return rc;	175,320✔
214	}
215
216	int
217	vy_point_lookup(struct vy_lsm lsm, struct vy_tx tx,	1,590,900✔
218	const struct vy_read_view **rv,
219	struct vy_entry key, struct vy_entry *ret)
220	{
221	/* All key parts must be set for a point lookup. */
222	assert(vy_stmt_is_full_key(key.stmt, lsm->cmp_def));	1,590,900!
223	assert(tx == NULL \|\| tx->state == VINYL_TX_READY);	1,590,900!
224
225	*ret = vy_entry_none();	1,590,900!
226	int rc = 0;	1,590,900✔
227
228	/* History list */
229	struct vy_history history, mem_history, disk_history;
230	vy_history_create(&history, &lsm->env->history_node_pool);	1,590,900!
231	vy_history_create(&mem_history, &lsm->env->history_node_pool);	1,590,900!
232	vy_history_create(&disk_history, &lsm->env->history_node_pool);	1,590,900!
233
234	rc = vy_point_lookup_scan_txw(lsm, tx, key, &history);	1,590,900!
235	if (rc != 0 \|\| vy_history_is_terminal(&history))	1,590,900!
236	goto done;	5,423✔
237
238	bool is_prepared_ok = tx != NULL ? vy_tx_is_prepared_ok(tx) : false;	1,585,470!
239	rc = vy_point_lookup_scan_cache(lsm, rv, is_prepared_ok, key, &history);	1,585,470!
240	if (rc != 0 \|\| vy_history_is_terminal(&history))	1,585,470!
241	goto done;	518,055✔
242
243	restart:	1,067,420✔
244	rc = vy_point_lookup_scan_mems(lsm, tx, rv, is_prepared_ok,	1,067,690!
245	key, &mem_history);
246	if (rc != 0 \|\| vy_history_is_terminal(&mem_history))	1,067,690!
247	goto done;	172,603✔
248
249	/* Save version before yield */
250	uint32_t mem_version = lsm->mem->version;	895,084✔
251	uint32_t mem_list_version = lsm->mem_list_version;	895,084✔
252
253	rc = vy_point_lookup_scan_slices(lsm, rv, key, &disk_history);	895,084!
254	if (rc != 0)	895,083!
UNCOV 255	goto done;	×
256
257	if (tx != NULL && tx->state == VINYL_TX_ABORT) {	895,083✔
258	/*
259	* The transaction was aborted while we were reading
260	* disk. We must stop now and return an error as this
261	* function could be called by a DML request aborted
262	* by a DDL operation: failing early will prevent it
263	* from dereferencing a destroyed space.
264	*/
265	diag_set(ClientError, ER_TRANSACTION_CONFLICT);	23!
266	rc = -1;	23✔
267	goto done;	23✔
268	}
269
270	if (mem_list_version != lsm->mem_list_version) {	895,060✔
271	/*
272	* Mem list was changed during yield. This could be rotation
273	* or a dump. In case of dump the memory referenced by
274	* statement history is gone and we need to reread new history.
275	* This in unnecessary in case of rotation but since we
276	* cannot distinguish these two cases we always restart.
277	*/
278	vy_history_cleanup(&mem_history);	269!
279	vy_history_cleanup(&disk_history);	269!
280	goto restart;	269✔
281	}
282
283	if (mem_version != lsm->mem->version) {	894,791✔
284	/*
285	* Rescan the memory level if its version changed while we
286	* were reading disk, because there may be new statements
287	* matching the search key.
288	*/
289	vy_history_cleanup(&mem_history);	23,747!
290	rc = vy_point_lookup_scan_mems(lsm, tx, rv, is_prepared_ok,	23,747!
291	key, &mem_history);
292	if (rc != 0)	23,747!
UNCOV 293	goto done;	×
294	if (vy_history_is_terminal(&mem_history))	23,747!
295	vy_history_cleanup(&disk_history);	1!
296	}
297
298	done:	894,790✔
299	vy_history_splice(&history, &mem_history);	1,590,900!
300	vy_history_splice(&history, &disk_history);	1,590,900!
301
302	if (rc == 0) {	1,590,900✔
303	int upserts_applied;
304	rc = vy_history_apply(&history, lsm->cmp_def,	1,590,870!
305	false, &upserts_applied, ret);
306	lsm->stat.upsert.applied += upserts_applied;	1,590,870✔
307	}
308	vy_history_cleanup(&history);	1,590,900!
309
310	if (rc != 0)	1,590,900✔
311	return -1;	1,590,900✔
312
313	return 0;	1,590,870✔
314	}
315
316	int
317	vy_point_lookup_mem(struct vy_lsm lsm, const struct vy_read_view *rv,	32,772✔
318	struct vy_entry key, struct vy_entry *ret)
319	{
320	assert(vy_stmt_is_full_key(key.stmt, lsm->cmp_def));	32,772!
321
322	int rc;
323	struct vy_history history;
324	vy_history_create(&history, &lsm->env->history_node_pool);	32,772!
325
326	rc = vy_point_lookup_scan_cache(lsm, rv, /is_prepared_ok=/true,	32,772!
327	key, &history);
328	if (rc != 0 \|\| vy_history_is_terminal(&history))	32,772!
329	goto done;	506✔
330
331	rc = vy_point_lookup_scan_mems(lsm, /tx=/NULL, rv,	32,266!
332	/is_prepared_ok=/true, key, &history);
333	if (rc != 0 \|\| vy_history_is_terminal(&history))	32,266!
334	goto done;	695✔
335
336	*ret = vy_entry_none();	31,571!
337	goto out;	31,571✔
338	done:	1,201✔
339	if (rc == 0) {	1,201!
340	int upserts_applied;
341	rc = vy_history_apply(&history, lsm->cmp_def,	1,201!
342	true, &upserts_applied, ret);
343	lsm->stat.upsert.applied += upserts_applied;	1,201✔
344	}
UNCOV 345	out:	×
346	vy_history_cleanup(&history);	32,772!
347	return rc;	32,772✔
348	}

tarantool / tarantool / 10992975672

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous