Skip to content

M
mariadb
Commit 8a273786 authored by marko's avatar marko
Browse files
Options

branches/5.1: Fix a bug in multi-table semi-consistent reads. 

Remember the acquired record locks per table handle (row_prebuilt_t)
rather than per transaction (trx_t), so that unlock_row should successfully
unlock all non-matching rows in multi-table operations.
This deficiency was found while investigating Bug #39320.

rb://94 approved by Heikki Tuuri.
parent 9387a126
Hide whitespace changes
Inline Side-by-side
Showing with 39 additions and 158 deletions
include/row0mysql.h
View file @ 8a273786
......@@ -656,6 +656,21 @@ struct row_prebuilt_struct {
This eliminates lock waits in some
cases; note that this breaks
serializability. */
ulint new_rec_locks; /* normally 0; if
srv_locks_unsafe_for_binlog is
TRUE or session is using READ
COMMITTED isolation level, in a
cursor search, if we set a new
record lock on an index, this is
incremented; this is used in
releasing the locks under the
cursors if we are performing an
UPDATE and we determine after
retrieving the row that it does
not need to be locked; thus,
these can be used to implement a
'mini-rollback' that releases
the latest record locks */
ulint mysql_prefix_len;/* byte offset of the end of
the last requested column */
ulint mysql_row_len; /* length in bytes of a row in the
......
include/trx0trx.h
View file @ 8a273786
......@@ -21,34 +21,6 @@ Created 3/26/1996 Heikki Tuuri
extern ulint trx_n_mysql_transactions;
/*****************************************************************
Resets the new record lock info in a transaction struct. */
UNIV_INLINE
void
trx_reset_new_rec_lock_info(
/*========================*/
trx_t* trx); /* in: transaction struct */
/*****************************************************************
Registers that we have set a new record lock on an index. We only have space
to store 2 indexes! If this is called to store more than 2 indexes after
trx_reset_new_rec_lock_info(), then this function does nothing. */
UNIV_INLINE
void
trx_register_new_rec_lock(
/*======================*/
trx_t* trx, /* in: transaction struct */
dict_index_t* index); /* in: trx sets a new record lock on this
index */
/*****************************************************************
Checks if trx has set a new record lock on an index. */
UNIV_INLINE
ibool
trx_new_rec_locks_contain(
/*======================*/
/* out: TRUE if trx has set a new record lock
on index */
trx_t* trx, /* in: transaction struct */
dict_index_t* index); /* in: index */
/************************************************************************
Releases the search latch if trx has reserved it. */
......@@ -527,20 +499,6 @@ struct trx_struct{
lock_t* auto_inc_lock; /* possible auto-inc lock reserved by
the transaction; note that it is also
in the lock list trx_locks */
dict_index_t* new_rec_locks[2];/* these are normally NULL; if
srv_locks_unsafe_for_binlog is TRUE
or session is using READ COMMITTED
isolation level,
in a cursor search, if we set a new
record lock on an index, this is set
to point to the index; this is
used in releasing the locks under the
cursors if we are performing an UPDATE
and we determine after retrieving
the row that it does not need to be
locked; thus, these can be used to
implement a 'mini-rollback' that
releases the latest record locks */
UT_LIST_NODE_T(trx_t)
trx_list; /* list of transactions */
UT_LIST_NODE_T(trx_t)
......
include/trx0trx.ic
View file @ 8a273786
......@@ -38,61 +38,3 @@ trx_start_if_not_started_low(
trx_start_low(trx, ULINT_UNDEFINED);
}
}
/*****************************************************************
Resets the new record lock info in a transaction struct. */
UNIV_INLINE
void
trx_reset_new_rec_lock_info(
/*========================*/
trx_t* trx) /* in: transaction struct */
{
trx->new_rec_locks[0] = NULL;
trx->new_rec_locks[1] = NULL;
}
/*****************************************************************
Registers that we have set a new record lock on an index. We only have space
to store 2 indexes! If this is called to store more than 2 indexes after
trx_reset_new_rec_lock_info(), then this function does nothing. */
UNIV_INLINE
void
trx_register_new_rec_lock(
/*======================*/
trx_t* trx, /* in: transaction struct */
dict_index_t* index) /* in: trx sets a new record lock on this
index */
{
if (trx->new_rec_locks[0] == NULL) {
trx->new_rec_locks[0] = index;
return;
}
if (trx->new_rec_locks[0] == index) {
return;
}
if (trx->new_rec_locks[1] != NULL) {
return;
}
trx->new_rec_locks[1] = index;
}
/*****************************************************************
Checks if trx has set a new record lock on an index. */
UNIV_INLINE
ibool
trx_new_rec_locks_contain(
/*======================*/
/* out: TRUE if trx has set a new record lock
on index */
trx_t* trx, /* in: transaction struct */
dict_index_t* index) /* in: index */
{
return(trx->new_rec_locks[0] == index
|| trx->new_rec_locks[1] == index);
}
lock/lock0lock.c
View file @ 8a273786
......@@ -1970,12 +1970,6 @@ lock_rec_lock_fast(
if (lock == NULL) {
if (!impl) {
lock_rec_create(mode, rec, index, trx);
if (srv_locks_unsafe_for_binlog
|| trx->isolation_level
== TRX_ISO_READ_COMMITTED) {
trx_register_new_rec_lock(trx, index);
}
}
return(TRUE);
......@@ -1999,11 +1993,6 @@ lock_rec_lock_fast(
if (!lock_rec_get_nth_bit(lock, heap_no)) {
lock_rec_set_nth_bit(lock, heap_no);
if (srv_locks_unsafe_for_binlog
|| trx->isolation_level
== TRX_ISO_READ_COMMITTED) {
trx_register_new_rec_lock(trx, index);
}
}
}
......@@ -2058,22 +2047,12 @@ lock_rec_lock_slow(
enough already granted on the record, we have to wait. */
err = lock_rec_enqueue_waiting(mode, rec, index, thr);
if (srv_locks_unsafe_for_binlog
|| trx->isolation_level == TRX_ISO_READ_COMMITTED) {
trx_register_new_rec_lock(trx, index);
}
} else {
if (!impl) {
/* Set the requested lock on the record */
lock_rec_add_to_queue(LOCK_REC | mode, rec, index,
trx);
if (srv_locks_unsafe_for_binlog
|| trx->isolation_level
== TRX_ISO_READ_COMMITTED) {
trx_register_new_rec_lock(trx, index);
}
}
err = DB_SUCCESS;
......
row/row0mysql.c
View file @ 8a273786
......@@ -1476,7 +1476,6 @@ row_unlock_for_mysql(
and clust_pcur, and we do not need to
reposition the cursors. */
{
dict_index_t* index;
btr_pcur_t* pcur = prebuilt->pcur;
btr_pcur_t* clust_pcur = prebuilt->clust_pcur;
trx_t* trx = prebuilt->trx;
......@@ -1501,9 +1500,7 @@ row_unlock_for_mysql(
trx->op_info = "unlock_row";
index = btr_pcur_get_btr_cur(pcur)->index;
if (index != NULL && trx_new_rec_locks_contain(trx, index)) {
if (prebuilt->new_rec_locks >= 1) {
mtr_start(&mtr);
......@@ -1518,22 +1515,9 @@ row_unlock_for_mysql(
lock_rec_unlock(trx, rec, prebuilt->select_lock_type);
mtr_commit(&mtr);
/* If the search was done through the clustered index, then
we have not used clust_pcur at all, and we must NOT try to
reset locks on clust_pcur. The values in clust_pcur may be
garbage! */
if (index->type & DICT_CLUSTERED) {
goto func_exit;
}
}
index = btr_pcur_get_btr_cur(clust_pcur)->index;
if (index != NULL && trx_new_rec_locks_contain(trx, index)) {
if (prebuilt->new_rec_locks >= 2) {
mtr_start(&mtr);
/* Restore the cursor position and find the record */
......@@ -1550,7 +1534,6 @@ row_unlock_for_mysql(
mtr_commit(&mtr);
}
func_exit:
trx->op_info = "";
return(DB_SUCCESS);
......
row/row0sel.c
View file @ 8a273786
......@@ -3304,13 +3304,7 @@ row_search_for_mysql(
is set or session is using a READ COMMITED isolation level. Then
we are able to remove the record locks set here on an individual
row. */
if ((srv_locks_unsafe_for_binlog
|| trx->isolation_level == TRX_ISO_READ_COMMITTED)
&& prebuilt->select_lock_type != LOCK_NONE) {
trx_reset_new_rec_lock_info(trx);
}
prebuilt->new_rec_locks = 0;
/*-------------------------------------------------------------*/
/* PHASE 1: Try to pop the row from the prefetch cache */
......@@ -3952,6 +3946,12 @@ no_gap_lock:
switch (err) {
rec_t* old_vers;
case DB_SUCCESS:
if (srv_locks_unsafe_for_binlog
|| trx->isolation_level == TRX_ISO_READ_COMMITTED) {
/* Note that a record of
prebuilt->index was locked. */
prebuilt->new_rec_locks = 1;
}
break;
case DB_LOCK_WAIT:
if (UNIV_LIKELY(prebuilt->row_read_type
......@@ -3982,7 +3982,7 @@ no_gap_lock:
if (UNIV_LIKELY(trx->wait_lock != NULL)) {
lock_cancel_waiting_and_release(
trx->wait_lock);
trx_reset_new_rec_lock_info(trx);
prebuilt->new_rec_locks = 0;
} else {
mutex_exit(&kernel_mutex);
......@@ -3994,6 +3994,9 @@ no_gap_lock:
ULINT_UNDEFINED,
&heap);
err = DB_SUCCESS;
/* Note that a record of
prebuilt->index was locked. */
prebuilt->new_rec_locks = 1;
break;
}
mutex_exit(&kernel_mutex);
......@@ -4143,6 +4146,15 @@ requires_clust_rec:
goto next_rec;
}
if ((srv_locks_unsafe_for_binlog
|| trx->isolation_level == TRX_ISO_READ_COMMITTED)
&& prebuilt->select_lock_type != LOCK_NONE) {
/* Note that both the secondary index record
and the clustered index record were locked. */
ut_ad(prebuilt->new_rec_locks == 1);
prebuilt->new_rec_locks = 2;
}
if (UNIV_UNLIKELY(rec_get_deleted_flag(clust_rec, comp))) {
/* The record is delete marked: we can skip it */
......@@ -4268,13 +4280,7 @@ next_rec:
prebuilt->row_read_type = ROW_READ_TRY_SEMI_CONSISTENT;
}
did_semi_consistent_read = FALSE;
if (UNIV_UNLIKELY(srv_locks_unsafe_for_binlog
|| trx->isolation_level == TRX_ISO_READ_COMMITTED)
&& prebuilt->select_lock_type != LOCK_NONE) {
trx_reset_new_rec_lock_info(trx);
}
prebuilt->new_rec_locks = 0;
/*-------------------------------------------------------------*/
/* PHASE 5: Move the cursor to the next index record */
......@@ -4380,7 +4386,7 @@ lock_wait_or_error:
rec_loop we will again try to set a lock, and
new_rec_lock_info in trx will be right at the end. */
trx_reset_new_rec_lock_info(trx);
prebuilt->new_rec_locks = 0;
}
mode = pcur->search_mode;
......
trx/trx0trx.c
View file @ 8a273786
......@@ -192,8 +192,6 @@ trx_create(
trx->n_autoinc_rows = 0;
trx_reset_new_rec_lock_info(trx);
return(trx);
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7