database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-08-01 03:47:19 +03:00
Code Activity

WL#3259 (RBR with more columns on slave than on master):

Browse Source 
Extended replication to allow extra columns added last on slave
as compared with table on master.
This commit is contained in:
mats@mysql.com
2006-05-03 15:00:38 +02:00
parent ed809462e9
commit 77c4a2ca60
11 changed files with 591 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
mysql-test/extra/rpl_tests/rpl_row_tabledefs.test
View File

@ -3,11 +3,16 @@
# Consider making these part of the basic RBR tests.
-- source include/have_binlog_format_row.inc
-- source include/master-slave.inc
connection slave;
STOP SLAVE;
SET GLOBAL SQL_MODE='STRICT_ALL_TABLES';
START SLAVE;
connection master;
eval CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t1_int (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t1_bit (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t1_char (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t1_nodef (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=$engine_type;
eval CREATE TABLE t4 (a INT) ENGINE=$engine_type;
@ -15,15 +20,21 @@ eval CREATE TABLE t5 (a INT, b INT, c INT) ENGINE=$engine_type;
eval CREATE TABLE t6 (a INT, b INT, c INT) ENGINE=$engine_type;
# Table used to detect that slave is running
eval CREATE TABLE t9 (a INT PRIMARY KEY) ENGINE=$engine_type;
eval CREATE TABLE t9 (a INT) ENGINE=$engine_type;
sync_slave_with_master;
# On the slave, we add one column last in table 't1',
ALTER TABLE t1 ADD x INT DEFAULT 42;
# ... add one column in the middle of table 't2', and
ALTER TABLE t2 ADD x INT DEFAULT 42 AFTER a;
# ... add one column first in table 't3'.
ALTER TABLE t3 ADD x INT DEFAULT 42 FIRST;
# On the slave, we add one INT column last in table 't1_int',
ALTER TABLE t1_int ADD x INT DEFAULT 42;
# ... and add one BIT column last in table 't1_bit',
ALTER TABLE t1_bit ADD x BIT(3) DEFAULT b'011';
# ... and add one CHAR column last in table 't1_char',
ALTER TABLE t1_char ADD x CHAR(20) DEFAULT 'Just a test';
# ... and add one non-nullable INT column last in table 't1_text'
# with no default,
ALTER TABLE t1_nodef ADD x INT NOT NULL;
# ... and remove the last column in t2
ALTER TABLE t2 DROP b;
# ... change the type of the single column in table 't4'
ALTER TABLE t4 MODIFY a FLOAT;
# ... change the type of the middle column of table 't5'
@ -31,13 +42,37 @@ ALTER TABLE t5 MODIFY b FLOAT;
# ... change the type of the last column of table 't6'
ALTER TABLE t6 MODIFY c FLOAT;
# Each of these should generate an error and stop the slave
# Insert some values for tables on slave side. These should not be
# modified when the row from the master is applied.
INSERT INTO t1_int VALUES (2,4,4711);
INSERT INTO t1_char VALUES (2,4,'Foo is a bar');
INSERT INTO t1_bit VALUES (2,4,b'101');
--echo **** On Master ****
connection master;
INSERT INTO t9 VALUES (1);
INSERT INTO t1_int VALUES (1,2);
INSERT INTO t1_int VALUES (2,5);
INSERT INTO t1_bit VALUES (1,2);
INSERT INTO t1_bit VALUES (2,5);
INSERT INTO t1_char VALUES (1,2);
INSERT INTO t1_char VALUES (2,5);
SELECT * FROM t1_int;
SELECT * FROM t1_bit;
SELECT * FROM t1_char;
--echo **** On Slave ****
sync_slave_with_master;
SELECT a,b,x FROM t1_int;
SELECT a,b,HEX(x) FROM t1_bit;
SELECT a,b,x FROM t1_char;
# Each of these should generate an error and stop the slave
connection master;
INSERT INTO t9 VALUES (2);
sync_slave_with_master;
# Now slave is guaranteed to be running
connection master;
INSERT INTO t1 VALUES (1,2);
INSERT INTO t1_nodef VALUES (1,2);
connection slave;
wait_for_slave_to_stop;
--replace_result $MASTER_MYPORT MASTER_PORT
@ -62,21 +97,6 @@ SHOW SLAVE STATUS;
SET GLOBAL SQL_SLAVE_SKIP_COUNTER=2;
START SLAVE;
connection master;
INSERT INTO t9 VALUES (3);
sync_slave_with_master;
# Now slave is guaranteed to be running
connection master;
INSERT INTO t3 VALUES (3,6);
connection slave;
wait_for_slave_to_stop;
--replace_result $MASTER_MYPORT MASTER_PORT
--replace_column 1 # 8 # 9 # 23 # 33 #
--vertical_results
SHOW SLAVE STATUS;
SET GLOBAL SQL_SLAVE_SKIP_COUNTER=2;
START SLAVE;
connection master;
INSERT INTO t9 VALUES (4);
sync_slave_with_master;
@ -124,6 +144,7 @@ START SLAVE;
connection master;
--disable_warnings
DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t9;
DROP TABLE IF EXISTS t1_int,t1_bit,t1_char,t1_nodef;
DROP TABLE IF EXISTS t2,t3,t4,t5,t6,t9;
--enable_warnings
sync_slave_with_master;

142
mysql-test/r/rpl_row_tabledefs.result
View File

@ -4,21 +4,64 @@ reset master;
reset slave;
drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
start slave;
CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
CREATE TABLE t4 (a INT) ENGINE=myisam;
CREATE TABLE t5 (a INT, b INT, c INT) ENGINE=myisam;
CREATE TABLE t6 (a INT, b INT, c INT) ENGINE=myisam;
CREATE TABLE t9 (a INT PRIMARY KEY) ENGINE=myisam;
ALTER TABLE t1 ADD x INT DEFAULT 42;
ALTER TABLE t2 ADD x INT DEFAULT 42 AFTER a;
ALTER TABLE t3 ADD x INT DEFAULT 42 FIRST;
STOP SLAVE;
SET GLOBAL SQL_MODE='STRICT_ALL_TABLES';
START SLAVE;
CREATE TABLE t1_int (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t1_bit (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t1_char (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t1_nodef (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE='MyISAM';
CREATE TABLE t4 (a INT) ENGINE='MyISAM';
CREATE TABLE t5 (a INT, b INT, c INT) ENGINE='MyISAM';
CREATE TABLE t6 (a INT, b INT, c INT) ENGINE='MyISAM';
CREATE TABLE t9 (a INT) ENGINE='MyISAM';
ALTER TABLE t1_int ADD x INT DEFAULT 42;
ALTER TABLE t1_bit ADD x BIT(3) DEFAULT b'011';
ALTER TABLE t1_char ADD x CHAR(20) DEFAULT 'Just a test';
ALTER TABLE t1_nodef ADD x INT NOT NULL;
ALTER TABLE t2 DROP b;
ALTER TABLE t4 MODIFY a FLOAT;
ALTER TABLE t5 MODIFY b FLOAT;
ALTER TABLE t6 MODIFY c FLOAT;
INSERT INTO t9 VALUES (1);
INSERT INTO t1 VALUES (1,2);
INSERT INTO t1_int VALUES (2,4,4711);
INSERT INTO t1_char VALUES (2,4,'Foo is a bar');
INSERT INTO t1_bit VALUES (2,4,b'101');
**** On Master ****
INSERT INTO t1_int VALUES (1,2);
INSERT INTO t1_int VALUES (2,5);
INSERT INTO t1_bit VALUES (1,2);
INSERT INTO t1_bit VALUES (2,5);
INSERT INTO t1_char VALUES (1,2);
INSERT INTO t1_char VALUES (2,5);
SELECT * FROM t1_int;
a b
1 2
2 5
SELECT * FROM t1_bit;
a b
1 2
2 5
SELECT * FROM t1_char;
a b
1 2
2 5
**** On Slave ****
SELECT a,b,x FROM t1_int;
a b x
2 5 4711
1 2 42
SELECT a,b,HEX(x) FROM t1_bit;
a b HEX(x)
2 5 5
1 2 3
SELECT a,b,x FROM t1_char;
a b x
2 5 Foo is a bar
1 2 Just a test
INSERT INTO t9 VALUES (2);
INSERT INTO t1_nodef VALUES (1,2);
SHOW SLAVE STATUS;
Slave_IO_State #
Master_Host 127.0.0.1
@ -26,7 +69,7 @@ Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1042
Read_Master_Log_Pos 1934
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
@ -38,10 +81,10 @@ Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Error Table width mismatch - received 2 columns, test.t1 has 3 columns
Last_Errno 1364
Last_Error Error in Write_rows event: error during transaction execution on table test.t1_nodef
Skip_Counter 0
Exec_Master_Log_Pos 968
Exec_Master_Log_Pos 1850
Relay_Log_Space #
Until_Condition None
Until_Log_File
@ -64,7 +107,7 @@ Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1185
Read_Master_Log_Pos 2085
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
@ -76,48 +119,10 @@ Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Error Table width mismatch - received 2 columns, test.t2 has 3 columns
Last_Errno 1514
Last_Error Table width mismatch - received 2 columns, test.t2 has 1 columns
Skip_Counter 0
Exec_Master_Log_Pos 1111
Relay_Log_Space #
Until_Condition None
Until_Log_File
Until_Log_Pos 0
Master_SSL_Allowed No
Master_SSL_CA_File
Master_SSL_CA_Path
Master_SSL_Cert
Master_SSL_Cipher
Master_SSL_Key
Seconds_Behind_Master #
SET GLOBAL SQL_SLAVE_SKIP_COUNTER=2;
START SLAVE;
INSERT INTO t9 VALUES (3);
INSERT INTO t3 VALUES (3,6);
SHOW SLAVE STATUS;
Slave_IO_State #
Master_Host 127.0.0.1
Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1328
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
Slave_IO_Running Yes
Slave_SQL_Running No
Replicate_Do_DB
Replicate_Ignore_DB
Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Error Table width mismatch - received 2 columns, test.t3 has 3 columns
Skip_Counter 0
Exec_Master_Log_Pos 1254
Exec_Master_Log_Pos 2007
Relay_Log_Space #
Until_Condition None
Until_Log_File
@ -140,7 +145,7 @@ Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1466
Read_Master_Log_Pos 2231
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
@ -152,10 +157,10 @@ Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Errno 1514
Last_Error Column 0 type mismatch - received type 3, test.t4 has type 4
Skip_Counter 0
Exec_Master_Log_Pos 1397
Exec_Master_Log_Pos 2158
Relay_Log_Space #
Until_Condition None
Until_Log_File
@ -178,7 +183,7 @@ Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1614
Read_Master_Log_Pos 2387
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
@ -190,10 +195,10 @@ Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Errno 1514
Last_Error Column 1 type mismatch - received type 3, test.t5 has type 4
Skip_Counter 0
Exec_Master_Log_Pos 1535
Exec_Master_Log_Pos 2304
Relay_Log_Space #
Until_Condition None
Until_Log_File
@ -216,7 +221,7 @@ Master_User root
Master_Port MASTER_PORT
Connect_Retry 1
Master_Log_File master-bin.000001
Read_Master_Log_Pos 1762
Read_Master_Log_Pos 2543
Relay_Log_File #
Relay_Log_Pos #
Relay_Master_Log_File master-bin.000001
@ -228,10 +233,10 @@ Replicate_Do_Table
Replicate_Ignore_Table
Replicate_Wild_Do_Table
Replicate_Wild_Ignore_Table
Last_Errno 1454
Last_Errno 1514
Last_Error Column 2 type mismatch - received type 3, test.t6 has type 4
Skip_Counter 0
Exec_Master_Log_Pos 1683
Exec_Master_Log_Pos 2460
Relay_Log_Space #
Until_Condition None
Until_Log_File
@ -245,4 +250,5 @@ Master_SSL_Key
Seconds_Behind_Master #
SET GLOBAL SQL_SLAVE_SKIP_COUNTER=2;
START SLAVE;
DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t9;
DROP TABLE IF EXISTS t1_int,t1_bit,t1_char,t1_nodef;
DROP TABLE IF EXISTS t2,t3,t4,t5,t6,t9;

8
mysql-test/t/rpl_row_tabledefs.test Normal file
View File

@ -0,0 +1,8 @@
-- source include/have_binlog_format_row.inc
-- source include/master-slave.inc
let $engine_type = 'MyISAM';
-- source extra/rpl_tests/rpl_row_tabledefs.test

4
sql/Makefile.am
View File

@ -53,7 +53,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \
sql_manager.h sql_map.h sql_string.h unireg.h \
sql_error.h field.h handler.h mysqld_suffix.h \
ha_heap.h ha_myisam.h ha_myisammrg.h ha_partition.h \
opt_range.h protocol.h rpl_tblmap.h \
opt_range.h protocol.h rpl_tblmap.h rpl_utility.h \
log.h sql_show.h rpl_rli.h \
sql_select.h structs.h table.h sql_udf.h hash_filo.h\
lex.h lex_symbol.h sql_acl.h sql_crypt.h \
@ -91,7 +91,7 @@ mysqld_SOURCES = sql_lex.cc sql_handler.cc sql_partition.cc \
sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
slave.cc sql_repl.cc rpl_filter.cc rpl_tblmap.cc \
rpl_injector.cc \
rpl_utility.cc rpl_injector.cc \
sql_union.cc sql_derived.cc \
client.c sql_client.cc mini_client_errors.c pack.c\
stacktrace.c repl_failsafe.h repl_failsafe.cc \

8
sql/field.cc
View File

@ -8259,6 +8259,14 @@ const char *Field_bit::unpack(char *to, const char *from)
}
void Field_bit::set_default()
{
my_ptrdiff_t const offset= table->s->default_values - table->record[0];
uchar bits= get_rec_bits(bit_ptr + offset, bit_ofs, bit_len);
set_rec_bits(bits, bit_ptr, bit_ofs, bit_len);
Field::set_default();
}
/*
Bit field support for non-MyISAM tables.
*/

2
sql/field.h
View File

@ -1384,6 +1384,8 @@ public:
void sql_type(String &str) const;
char *pack(char *to, const char *from, uint max_length=~(uint) 0);
const char *unpack(char* to, const char *from);
virtual void set_default();
Field *new_key_field(MEM_ROOT *root, struct st_table *new_table,
char *new_ptr, uchar *new_null_ptr,
uint new_null_bit);

327
sql/log_event.cc
View File

@ -24,6 +24,7 @@
#include "mysql_priv.h"
#include "slave.h"
#include "rpl_filter.h"
#include "rpl_utility.h"
#include <my_dir.h>
#endif /* MYSQL_CLIENT */
#include <base64.h>
@ -5258,38 +5259,86 @@ int Rows_log_event::do_add_row_data(byte *const row_data,
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
/*
Unpack a row into a record. The row is assumed to only consist of the fields
for which the bitset represented by 'arr' and 'bits'; the other parts of the
Unpack a row into a record.
The row is assumed to only consist of the fields for which the
bitset represented by 'arr' and 'bits'; the other parts of the
record are left alone.
At most 'colcnt' columns are read: if the table is larger than that,
the remaining fields are not filled in.
*/
static char const *unpack_row(TABLE *table,
byte *record, char const *row,
MY_BITMAP const *cols)
static int
unpack_row(RELAY_LOG_INFO *rli,
TABLE *table, uint colcnt, byte *record,
char const *row, MY_BITMAP const *cols,
char const **row_end, ulong *master_reclength)
{
DBUG_ASSERT(record && row);
MY_BITMAP *write_set= table->file->write_set;
my_size_t const n_null_bytes= table->s->null_bytes;
my_ptrdiff_t const offset= record - (byte*) table->record[0];
memcpy(record, row, n_null_bytes);
char const *ptr= row + n_null_bytes;
memcpy(record, row, n_null_bytes); // [1]
int error= 0;
bitmap_set_all(write_set);
Field **const begin_ptr = table->field;
for (Field **field_ptr= begin_ptr ; *field_ptr ; ++field_ptr)
{
Field *const f= *field_ptr;
if (bitmap_is_set(cols, field_ptr - begin_ptr))
Field **const begin_ptr = table->field;
Field **field_ptr;
{
char const *ptr= row + n_null_bytes;
for (field_ptr= begin_ptr ; *field_ptr ; ++field_ptr)
{
/* Field...::unpack() cannot return 0 */
ptr= f->unpack(f->ptr + offset, ptr);
Field *const f= *field_ptr;
if (colcnt == 0)
break;
if (bitmap_is_set(cols, field_ptr - begin_ptr))
{
/* Field...::unpack() cannot return 0 */
ptr= f->unpack(f->ptr + offset, ptr);
--colcnt;
}
else
bitmap_clear_bit(write_set, (field_ptr - begin_ptr) + 1);
}
*row_end = ptr;
if (master_reclength)
{
if (*field_ptr)
*master_reclength = (*field_ptr)->ptr - table->record[0];
else
*master_reclength = table->s->reclength;
}
}
/*
Set properties for remaining columns, if there are any. We let the
corresponding bit in the write_set be set, to write the value if
it was not there already. We iterate over all remaining columns,
even if there were an error, to get as many error messages as
possible. We are still able to return a pointer to the next row,
so wedo that.
*/
for ( ; *field_ptr ; ++field_ptr)
{
if ((*field_ptr)->flags & (NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG))
{
slave_print_msg(ERROR_LEVEL, rli, ER_NO_DEFAULT_FOR_FIELD,
"Field `%s` of table `%s`.`%s` "
"has no default value and cannot be NULL",
(*field_ptr)->field_name, table->s->db.str,
table->s->table_name.str);
error = ER_NO_DEFAULT_FOR_FIELD;
}
else
bitmap_clear_bit(write_set, (field_ptr - begin_ptr) + 1);
(*field_ptr)->set_default();
}
return ptr;
return error;
}
int Rows_log_event::exec_event(st_relay_log_info *rli)
@ -5444,7 +5493,11 @@ int Rows_log_event::exec_event(st_relay_log_info *rli)
error= do_before_row_operations(table);
while (error == 0 && row_start < (const char*)m_rows_end) {
char const *row_end= do_prepare_row(thd, table, row_start);
char const *row_end= NULL;
if ((error= do_prepare_row(thd, rli, table, row_start, &row_end)))
break; // We should to the after-row operation even in the
// case of error
DBUG_ASSERT(row_end != NULL); // cannot happen
DBUG_ASSERT(row_end <= (const char*)m_rows_end);
@ -5646,7 +5699,7 @@ void Rows_log_event::pack_info(Protocol *protocol)
#endif
/**************************************************************************
Table_map_log_event member functions
Table_map_log_event member functions and support functions
**************************************************************************/
/*
@ -5910,71 +5963,9 @@ int Table_map_log_event::exec_event(st_relay_log_info *rli)
*/
DBUG_ASSERT(m_table->in_use);
/*
Check that the number of columns and the field types in the
event match the number of columns and field types in the opened
table.
*/
uint col= m_table->s->fields;
if (col == m_colcnt)
table_def const def(m_coltype, m_colcnt);
if (def.compatible_with(rli, m_table))
{
while (col-- > 0)
if (m_table->field[col]->type() != m_coltype[col])
break;
}
TABLE_SHARE const *const tsh= m_table->s;
/*
Check the following termination conditions:
(col == m_table->s->fields)
==> (m_table->s->fields != m_colcnt)
(0 <= col < m_table->s->fields)
==> (m_table->field[col]->type() != m_coltype[col])
Logically, A ==> B is equivalent to !A || B
Since col is unsigned, is suffices to check that col <=
tsh->fields. If col wrapped (by decreasing col when it is 0),
the number will be UINT_MAX, which is greater than tsh->fields.
*/
DBUG_ASSERT(!(col == tsh->fields) || tsh->fields != m_colcnt);
DBUG_ASSERT(!(col < tsh->fields) ||
(m_table->field[col]->type() != m_coltype[col]));
if (col <= tsh->fields)
{
/*
If we get here, the number of columns in the event didn't
match the number of columns in the table on the slave, *or*
there were a column in the table on the slave that did not
have the same type as given in the event.
If 'col' has the value that was assigned to it, it was a
mismatch between the number of columns on the master and the
slave.
*/
if (col == tsh->fields)
{
DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
"Table width mismatch - "
"received %u columns, %s.%s has %u columns",
m_colcnt, tsh->db.str, tsh->table_name.str, tsh->fields);
}
else
{
DBUG_ASSERT(col < m_colcnt && col < tsh->fields);
DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
"Column %d type mismatch - "
"received type %d, %s.%s has type %d",
col, m_coltype[col], tsh->db.str, tsh->table_name.str,
m_table->field[col]->type());
}
thd->query_error= 1;
DBUG_RETURN(ERR_BAD_TABLE_DEF);
}
@ -6085,6 +6076,25 @@ void Table_map_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info)
}
#endif
#ifndef DBUG_OFF
static void
print_column_values(char const *text, THD *thd, TABLE *table)
{
THD *old_thd= table->in_use;
if (table->in_use == NULL)
table->in_use= thd;
for (Field **fptr= table->field ; *fptr ; ++fptr)
{
char buf[MAX_FIELD_WIDTH];
String str(buf, sizeof(buf), system_charset_info);
(*fptr)->val_str(&str);
DBUG_PRINT("info", ("%s for column %d is '%s'",
text, fptr - table->field, str.c_ptr()));
}
table->in_use= old_thd;
}
#endif
/**************************************************************************
Write_rows_log_event member functions
**************************************************************************/
@ -6169,19 +6179,22 @@ int Write_rows_log_event::do_after_row_operations(TABLE *table, int error)
return error;
}
char const *Write_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
char const *row_start)
int Write_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
TABLE *table,
char const *row_start,
char const **row_end)
{
char const *ptr= row_start;
DBUG_ASSERT(table != NULL);
/*
This assertion actually checks that there is at least as many
columns on the slave as on the master.
*/
DBUG_ASSERT(table->s->fields >= m_width);
DBUG_ASSERT(ptr);
ptr= unpack_row(table, (byte*)table->record[0], ptr, &m_cols);
return ptr;
DBUG_ASSERT(row_start && row_end);
int error;
error= unpack_row(rli,
table, m_width, (byte*)table->record[0],
row_start, &m_cols, row_end, &m_master_reclength);
#ifndef DBUG_OFF
print_column_values("Unpacked value", thd, table);
#endif
return error;
}
/*
@ -6237,24 +6250,33 @@ namespace {
@param thd Thread context for writing the record.
@param table Table to which record should be written.
@param master_reclength
Offset to first column that is not present on the master,
alternatively the length of the record on the master side.
@return Error code on failure, 0 on success.
*/
static int
replace_record(THD *thd, TABLE *table)
replace_record(THD *thd, TABLE *table,
ulong const master_reclength,
uint const master_fields)
{
DBUG_ENTER("replace_record");
DBUG_ASSERT(table != NULL && thd != NULL);
int error;
int keynum;
auto_afree_ptr<char> key(NULL);
#ifndef DBUG_OFF
print_column_values("Starting write value", thd, table);
#endif
while ((error= table->file->ha_write_row(table->record[0])))
{
if ((keynum= table->file->get_dup_key(error)) < 0)
{
/* We failed to retrieve the duplicate key */
return HA_ERR_FOUND_DUPP_KEY;
DBUG_RETURN(HA_ERR_FOUND_DUPP_KEY);
}
/*
@ -6271,20 +6293,20 @@ replace_record(THD *thd, TABLE *table)
{
error= table->file->rnd_pos(table->record[1], table->file->dupp_ref);
if (error)
return error;
DBUG_RETURN(error);
}
else
{
if (table->file->extra(HA_EXTRA_FLUSH_CACHE))
{
return my_errno;
DBUG_RETURN(my_errno);
}
if (key.get() == NULL)
{
key.assign(static_cast<char*>(my_alloca(table->s->max_unique_length)));
if (key.get() == NULL)
return ENOMEM;
DBUG_RETURN(ENOMEM);
}
key_copy((byte*)key.get(), table->record[0], table->key_info + keynum, 0);
@ -6293,7 +6315,7 @@ replace_record(THD *thd, TABLE *table)
table->key_info[keynum].key_length,
HA_READ_KEY_EXACT);
if (error)
return error;
DBUG_RETURN(error);
}
/*
@ -6301,6 +6323,59 @@ replace_record(THD *thd, TABLE *table)
will enable us to update it or, alternatively, delete it (so
that we can insert the new row afterwards).
First we copy the columns into table->record[0] that are not
present on the master from table->record[1], if there are any.
*/
DBUG_PRINT("info", ("Copying to %p from offset %u to %u",
table->record[0],
master_reclength, table->s->reclength));
#ifndef DBUG_OFF
print_column_values("After copy value", thd, table);
#endif
if (master_reclength < table->s->reclength)
bmove_align(table->record[0] + master_reclength,
table->record[1] + master_reclength,
table->s->reclength - master_reclength);
/*
Bit columns are special. We iterate over all the remaining
columns and copy the "extra" bits to the new record. This is
not a very good solution: it should be refactored on
opportunity.
REFACTORING SUGGESTION (Matz). Introduce a member function
similar to move_field_offset() called copy_field_offset() to
copy field values and implement it for all Field subclasses. Use
this function to copy data from the found record to the record
that are going to be inserted.
The copy_field_offset() function need to be a virtual function,
which in this case will prevent copying an entire range of
fields efficiently.
*/
{
Field **field_ptr= table->field + master_fields;
for ( ; *field_ptr ; ++field_ptr)
{
switch ((*field_ptr)->real_type())
{
default:
/* Nothing to do */
break;
case FIELD_TYPE_BIT:
Field_bit *f= static_cast<Field_bit*>(*field_ptr);
my_ptrdiff_t const offset= table->record[1] - table->record[0];
uchar const bits=
get_rec_bits(f->bit_ptr + offset, f->bit_ofs, f->bit_len);
set_rec_bits(bits, f->bit_ptr, f->bit_ofs, f->bit_len);
break;
}
}
}
/*
REPLACE is defined as either INSERT or DELETE + INSERT. If
possible, we can replace it with an UPDATE, but that will not
work on InnoDB if FOREIGN KEY checks are necessary.
@ -6320,22 +6395,22 @@ replace_record(THD *thd, TABLE *table)
{
error=table->file->ha_update_row(table->record[1],
table->record[0]);
return error;
DBUG_RETURN(error);
}
else
{
if ((error= table->file->ha_delete_row(table->record[1])))
return error;
DBUG_RETURN(error);
/* Will retry ha_write_row() with the offending row removed. */
}
}
return error;
DBUG_RETURN(error);
}
int Write_rows_log_event::do_exec_row(TABLE *table)
{
DBUG_ASSERT(table != NULL);
int error= replace_record(thd, table);
int error= replace_record(thd, table, m_master_reclength, m_width);
return error;
}
#endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */
@ -6606,20 +6681,22 @@ int Delete_rows_log_event::do_after_row_operations(TABLE *table, int error)
return error;
}
char const *Delete_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
char const *row_start)
int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
TABLE *table,
char const *row_start,
char const **row_end)
{
char const *ptr= row_start;
DBUG_ASSERT(ptr);
int error;
DBUG_ASSERT(row_start && row_end);
/*
This assertion actually checks that there is at least as many
columns on the slave as on the master.
*/
DBUG_ASSERT(table->s->fields >= m_width);
DBUG_ASSERT(ptr != NULL);
ptr= unpack_row(table, table->record[0], ptr, &m_cols);
error= unpack_row(rli,
table, m_width, table->record[0],
row_start, &m_cols, row_end, &m_master_reclength);
/*
If we will access rows using the random access method, m_key will
be set to NULL, so we do not need to make a key copy in that case.
@ -6631,7 +6708,7 @@ char const *Delete_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
key_copy(m_key, table->record[0], key_info, 0);
}
return ptr;
return error;
}
int Delete_rows_log_event::do_exec_row(TABLE *table)
@ -6757,11 +6834,13 @@ int Update_rows_log_event::do_after_row_operations(TABLE *table, int error)
return error;
}
char const *Update_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
char const *row_start)
int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
TABLE *table,
char const *row_start,
char const **row_end)
{
char const *ptr= row_start;
DBUG_ASSERT(ptr);
int error;
DBUG_ASSERT(row_start && row_end);
/*
This assertion actually checks that there is at least as many
columns on the slave as on the master.
@ -6769,10 +6848,14 @@ char const *Update_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
DBUG_ASSERT(table->s->fields >= m_width);
/* record[0] is the before image for the update */
ptr= unpack_row(table, table->record[0], ptr, &m_cols);
DBUG_ASSERT(ptr != NULL);
error= unpack_row(rli,
table, m_width, table->record[0],
row_start, &m_cols, row_end, &m_master_reclength);
row_start = *row_end;
/* m_after_image is the after image for the update */
ptr= unpack_row(table, m_after_image, ptr, &m_cols);
error= unpack_row(rli,
table, m_width, m_after_image,
row_start, &m_cols, row_end, &m_master_reclength);
/*
If we will access rows using the random access method, m_key will
@ -6785,7 +6868,7 @@ char const *Update_rows_log_event::do_prepare_row(THD *thd, TABLE *table,
key_copy(m_key, table->record[0], key_info, 0);
}
return ptr;
return error;
}
int Update_rows_log_event::do_exec_row(TABLE *table)

35
sql/log_event.h
View File

@ -1854,6 +1854,7 @@ protected:
ulong m_table_id; /* Table ID */
MY_BITMAP m_cols; /* Bitmap denoting columns available */
ulong m_width; /* The width of the columns bitmap */
ulong m_master_reclength; /* Length of record on master side */
/* Bit buffer in the same memory as the class */
uint32 m_bitbuf[128/(sizeof(uint32)*8)];
@ -1907,12 +1908,15 @@ private:
since SQL thread specific data is not available: that data is made
available for the do_exec function.
RETURN VALUE
A pointer to the start of the next row, or NULL if the preparation
failed. Currently, preparation cannot fail, but don't rely on this
behavior.
RETURN VALUE
Error code, if something went wrong, 0 otherwise.
*/
virtual char const *do_prepare_row(THD*, TABLE*, char const *row_start) = 0;
virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
char const *row_start, char const **row_end) = 0;
/*
Primitive to do the actual execution necessary for a row.
@ -1980,10 +1984,11 @@ private:
gptr m_memory;
byte *m_after_image;
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual char const *do_prepare_row(THD*, TABLE*, char const *row_start);
virtual int do_exec_row(TABLE *table);
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
char const *row_start, char const **row_end);
virtual int do_exec_row(TABLE *table);
#endif
};
@ -2044,10 +2049,11 @@ private:
byte *m_key;
byte *m_after_image;
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual char const *do_prepare_row(THD*, TABLE*, char const *row_start);
virtual int do_exec_row(TABLE *table);
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
char const *row_start, char const **row_end);
virtual int do_exec_row(TABLE *table);
#endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */
};
@ -2114,10 +2120,11 @@ private:
byte *m_key;
byte *m_after_image;
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual char const *do_prepare_row(THD*, TABLE*, char const *row_start);
virtual int do_exec_row(TABLE *table);
virtual int do_before_row_operations(TABLE *table);
virtual int do_after_row_operations(TABLE *table, int error);
virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
char const *row_start, char const **row_end);
virtual int do_exec_row(TABLE *table);
#endif
};

5
sql/mysql_priv.h
View File

@ -21,6 +21,9 @@
except the part which must be in the server and in the client.
*/
#ifndef MYSQL_PRIV_H
#define MYSQL_PRIV_H
#ifndef MYSQL_CLIENT
#include <my_global.h>
@ -1773,3 +1776,5 @@ bool schema_table_store_record(THD *thd, TABLE *table);
#endif /* MYSQL_SERVER */
#endif /* MYSQL_CLIENT */
#endif /* MYSQL_PRIV_H */

156
sql/rpl_utility.cc Normal file
View File

@ -0,0 +1,156 @@
/* Copyright 2006 MySQL AB. All rights reserved.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "rpl_utility.h"
uint32
field_length_from_packed(enum_field_types const field_type,
byte const *const data)
{
uint32 length;
switch (field_type) {
case MYSQL_TYPE_DECIMAL:
case MYSQL_TYPE_NEWDECIMAL:
length= ~0UL;
break;
case MYSQL_TYPE_YEAR:
case MYSQL_TYPE_TINY:
length= 1;
break;
case MYSQL_TYPE_SHORT:
length= 2;
break;
case MYSQL_TYPE_INT24:
length= 3;
break;
case MYSQL_TYPE_LONG:
length= 4;
break;
#ifdef HAVE_LONG_LONG
case MYSQL_TYPE_LONGLONG:
length= 8;
break;
#endif
case MYSQL_TYPE_FLOAT:
length= sizeof(float);
break;
case MYSQL_TYPE_DOUBLE:
length= sizeof(double);
break;
case MYSQL_TYPE_NULL:
length= 0;
break;
case MYSQL_TYPE_NEWDATE:
length= 3;
break;
case MYSQL_TYPE_DATE:
length= 4;
break;
case MYSQL_TYPE_TIME:
length= 3;
break;
case MYSQL_TYPE_TIMESTAMP:
length= 4;
break;
case MYSQL_TYPE_DATETIME:
length= 8;
break;
break;
case MYSQL_TYPE_BIT:
length= ~0UL;
break;
default:
/* This case should never be chosen */
DBUG_ASSERT(0);
/* If something goes awfully wrong, it's better to get a string than die */
case MYSQL_TYPE_STRING:
length= uint2korr(data);
break;
case MYSQL_TYPE_ENUM:
case MYSQL_TYPE_SET:
case MYSQL_TYPE_VAR_STRING:
case MYSQL_TYPE_VARCHAR:
length= ~0UL; // NYI
break;
case MYSQL_TYPE_TINY_BLOB:
case MYSQL_TYPE_MEDIUM_BLOB:
case MYSQL_TYPE_LONG_BLOB:
case MYSQL_TYPE_BLOB:
case MYSQL_TYPE_GEOMETRY:
length= ~0UL; // NYI
break;
}
}
/*********************************************************************
* table_def member definitions *
*********************************************************************/
/*
Is the definition compatible with a table?
Compare the definition with a table to see if it is compatible with
it. A table definition is compatible with a table if
- the columns types of the table definition is a (not necessarily
proper) prefix of the column type of the table, or
- the other way around
*/
int
table_def::compatible_with(RELAY_LOG_INFO *rli, TABLE *table)
const
{
/*
We only check the initial columns for the tables.
*/
uint const cols_to_check= min(table->s->fields, size());
int error= 0;
TABLE_SHARE const *const tsh= table->s;
/*
To get proper error reporting for all columns of the table, we
both check the width and iterate over all columns.
*/
if (tsh->fields < size())
{
DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
error= 1;
slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
"Table width mismatch - "
"received %u columns, %s.%s has %u columns",
size(), tsh->db.str, tsh->table_name.str, tsh->fields);
}
for (uint col= 0 ; col < cols_to_check ; ++col)
{
if (table->field[col]->type() != type(col))
{
DBUG_ASSERT(col < size() && col < tsh->fields);
DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
error= 1;
slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
"Column %d type mismatch - "
"received type %d, %s.%s has type %d",
col, type(col), tsh->db.str, tsh->table_name.str,
table->field[col]->type());
}
}
return error;
}

60
sql/rpl_utility.h Normal file
View File

@ -0,0 +1,60 @@
/* Copyright 2006 MySQL AB. All rights reserved.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#ifndef RPL_UTILITY_H
#define RPL_UTILITY_H
#ifndef __cplusplus
#error "Don't include this C++ header file from a non-C++ file!"
#endif
#include "mysql_priv.h"
uint32
field_length_from_packed(enum_field_types const field_type,
byte const *const data);
/*
A table definition from the master.
RESPONSIBILITIES
- Extract table definition data from the table map event
- Check if table definition in table map is compatible with table
definition on slave
*/
class table_def
{
public:
typedef unsigned char field_type;
table_def(field_type *t, my_size_t s)
: m_type(t), m_size(s)
{
}
my_size_t size() const { return m_size; }
field_type type(my_ptrdiff_t i) const { return m_type[i]; }
int compatible_with(RELAY_LOG_INFO *rli, TABLE *table) const;
private:
my_size_t m_size;
field_type *m_type;
};
#endif /* RPL_UTILITY_H */