diff options
Diffstat (limited to 'sql/log_event_server.cc')
| -rw-r--r-- | sql/log_event_server.cc | 52 |
1 files changed, 43 insertions, 9 deletions
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc index 013515850ef..f80cf68623a 100644 --- a/sql/log_event_server.cc +++ b/sql/log_event_server.cc @@ -7543,8 +7543,18 @@ Rows_log_event::write_row(rpl_group_info *rgi, int Rows_log_event::update_sequence() { TABLE *table= m_table; // pointer to event's table + bool old_master= false; + int err= 0; - if (!bitmap_is_set(table->rpl_write_set, MIN_VALUE_FIELD_NO)) + if (!bitmap_is_set(table->rpl_write_set, MIN_VALUE_FIELD_NO) || + ( +#if defined(WITH_WSREP) + ! WSREP(thd) && +#endif + !(table->in_use->rgi_slave->gtid_ev_flags2 & Gtid_log_event::FL_DDL) && + !(old_master= + rpl_master_has_bug(thd->rgi_slave->rli, + 29621, FALSE, FALSE, FALSE, TRUE)))) { /* This event come from a setval function executed on the master. Update the sequence next_number and round, like we do with setval() @@ -7557,12 +7567,27 @@ int Rows_log_event::update_sequence() return table->s->sequence->set_value(table, nextval, round, 0) > 0; } - + if (old_master && !WSREP(thd) && thd->rgi_slave->is_parallel_exec) + { + DBUG_ASSERT(thd->rgi_slave->parallel_entry); + /* + With parallel replication enabled, we can't execute alongside any other + transaction in which we may depend, so we force retry to release + the server layer table lock for possible prior in binlog order + same table transactions. + */ + if (thd->rgi_slave->parallel_entry->last_committed_sub_id < + thd->rgi_slave->wait_commit_sub_id) + { + err= ER_LOCK_DEADLOCK; + my_error(err, MYF(0)); + } + } /* Update all fields in table and update the active sequence, like with ALTER SEQUENCE */ - return table->file->ha_write_row(table->record[0]); + return err == 0 ? table->file->ha_write_row(table->record[0]) : err; } @@ -7631,7 +7656,7 @@ uint8 Write_rows_log_event::get_trg_event_map() Returns TRUE if different. */ -static bool record_compare(TABLE *table) +static bool record_compare(TABLE *table, bool vers_from_plain= false) { bool result= FALSE; /** @@ -7664,10 +7689,19 @@ static bool record_compare(TABLE *table) /* Compare fields */ for (Field **ptr=table->field ; *ptr ; ptr++) { - if (table->versioned() && (*ptr)->vers_sys_field()) - { + /* + If the table is versioned, don't compare using the version if there is a + primary key. If there isn't a primary key, we need the version to + identify the correct record if there are duplicate rows in the data set. + However, if the primary server is unversioned (vers_from_plain is true), + then we implicitly use row_end as the primary key on our side. This is + because the implicit row_end value will be set to the maximum value for + the latest row update (which is what we care about). + */ + if (table->versioned() && (*ptr)->vers_sys_field() && + (table->s->primary_key < MAX_KEY || + (vers_from_plain && table->vers_start_field() == (*ptr)))) continue; - } /** We only compare field contents that are not null. NULL fields (i.e., their null bits) were compared @@ -8064,7 +8098,7 @@ int Rows_log_event::find_row(rpl_group_info *rgi) /* We use this to test that the correct key is used in test cases. */ DBUG_EXECUTE_IF("slave_crash_if_index_scan", abort();); - while (record_compare(table)) + while (record_compare(table, m_vers_from_plain)) { while ((error= table->file->ha_index_next(table->record[0]))) { @@ -8117,7 +8151,7 @@ int Rows_log_event::find_row(rpl_group_info *rgi) goto end; } } - while (record_compare(table)); + while (record_compare(table, m_vers_from_plain)); /* Note: above record_compare will take into accout all record fields |