diff options
Diffstat (limited to 'sql')
56 files changed, 2738 insertions, 1002 deletions
diff --git a/sql/Makefile.am b/sql/Makefile.am index ac14ed3e883..a85eb012f1d 100644 --- a/sql/Makefile.am +++ b/sql/Makefile.am @@ -52,7 +52,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \ sql_error.h field.h handler.h mysqld_suffix.h \ ha_partition.h \ ha_ndbcluster.h ha_ndbcluster_binlog.h \ - ha_ndbcluster_tables.h \ + ha_ndbcluster_tables.h rpl_constants.h \ opt_range.h protocol.h rpl_tblmap.h rpl_utility.h \ log.h sql_show.h rpl_rli.h rpl_mi.h \ sql_select.h structs.h table.h sql_udf.h hash_filo.h \ diff --git a/sql/event_data_objects.cc b/sql/event_data_objects.cc index cc8768bc58f..b65539b5aea 100644 --- a/sql/event_data_objects.cc +++ b/sql/event_data_objects.cc @@ -101,7 +101,8 @@ Event_parse_data::new_instance(THD *thd) */ Event_parse_data::Event_parse_data() - :on_completion(ON_COMPLETION_DROP), status(ENABLED), do_not_create(FALSE), + :on_completion(Event_basic::ON_COMPLETION_DROP), + status(Event_basic::ENABLED), do_not_create(FALSE), item_starts(NULL), item_ends(NULL), item_execute_at(NULL), starts_null(TRUE), ends_null(TRUE), execute_at_null(TRUE), item_expression(NULL), expression(0) @@ -243,7 +244,7 @@ Event_parse_data::check_if_in_the_past(THD *thd, my_time_t ltime_utc) if (ltime_utc >= (my_time_t) thd->query_start()) return; - if (on_completion == ON_COMPLETION_DROP) + if (on_completion == Event_basic::ON_COMPLETION_DROP) { switch (thd->lex->sql_command) { case SQLCOM_CREATE_EVENT: @@ -260,9 +261,9 @@ Event_parse_data::check_if_in_the_past(THD *thd, my_time_t ltime_utc) do_not_create= TRUE; } - else if (status == ENABLED) + else if (status == Event_basic::ENABLED) { - status= DISABLED; + status= Event_basic::DISABLED; push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_EVENT_EXEC_TIME_IN_THE_PAST, ER(ER_EVENT_EXEC_TIME_IN_THE_PAST)); @@ -588,9 +589,9 @@ Event_parse_data::check_parse_data(THD *thd) init_name(thd, identifier); init_definer(thd); - ret= init_execute_at(thd) || init_interval(thd) || init_starts(thd) || init_ends(thd); + check_originator_id(thd); DBUG_RETURN(ret); } @@ -638,6 +639,31 @@ Event_parse_data::init_definer(THD *thd) /* + Set the originator id of the event to the server_id if executing on + the master or set to the server_id of the master if executing on + the slave. If executing on slave, also set status to SLAVESIDE_DISABLED. + + SYNOPSIS + Event_parse_data::check_originator_id() +*/ +void Event_parse_data::check_originator_id(THD *thd) +{ + /* Disable replicated events on slave. */ + if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL) || + (thd->system_thread == SYSTEM_THREAD_SLAVE_IO)) + { + DBUG_PRINT("info", ("Invoked object status set to SLAVESIDE_DISABLED.")); + if ((status == Event_basic::ENABLED) || + (status == Event_basic::DISABLED)) + status = Event_basic::SLAVESIDE_DISABLED; + originator = thd->server_id; + } + else + originator = server_id; +} + + +/* Constructor SYNOPSIS @@ -995,8 +1021,23 @@ Event_queue_element::load_from_row(THD *thd, TABLE *table) goto error; DBUG_PRINT("load_from_row", ("Event [%s] is [%s]", name.str, ptr)); - status= (ptr[0]=='E'? Event_queue_element::ENABLED: - Event_queue_element::DISABLED); + + /* Set event status (ENABLED | SLAVESIDE_DISABLED | DISABLED) */ + switch (ptr[0]) + { + case 'E' : + status = Event_queue_element::ENABLED; + break; + case 'S' : + status = Event_queue_element::SLAVESIDE_DISABLED; + break; + case 'D' : + status = Event_queue_element::DISABLED; + break; + } + if ((ptr= get_field(&mem_root, table->field[ET_FIELD_ORIGINATOR])) == NullS) + goto error; + originator = table->field[ET_FIELD_ORIGINATOR]->val_int(); /* ToDo : Andrey . Find a way not to allocate ptr on event_mem_root */ if ((ptr= get_field(&mem_root, @@ -1343,7 +1384,7 @@ Event_queue_element::compute_next_execution_time() (long) starts, (long) ends, (long) last_executed, (long) this)); - if (status == Event_queue_element::DISABLED) + if (status != Event_queue_element::ENABLED) { DBUG_PRINT("compute_next_execution_time", ("Event %s is DISABLED", name.str)); @@ -1641,6 +1682,13 @@ Event_queue_element::update_timing_fields(THD *thd) status_changed= FALSE; } + /* + Turn off row binlogging of event timing updates. These are not used + for RBR of events replicated to the slave. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + if ((table->file->ha_update_row(table->record[1], table->record[0]))) ret= TRUE; @@ -1731,6 +1779,8 @@ Event_timed::get_create_event(THD *thd, String *buf) if (status == Event_timed::ENABLED) buf->append(STRING_WITH_LEN("ENABLE")); + else if (status == Event_timed::SLAVESIDE_DISABLED) + buf->append(STRING_WITH_LEN("SLAVESIDE_DISABLE")); else buf->append(STRING_WITH_LEN("DISABLE")); diff --git a/sql/event_data_objects.h b/sql/event_data_objects.h index 8e013e40400..d1d213901f0 100644 --- a/sql/event_data_objects.h +++ b/sql/event_data_objects.h @@ -22,11 +22,9 @@ #define EVEX_BAD_PARAMS -5 #define EVEX_MICROSECOND_UNSUP -6 - class sp_head; class Sql_alloc; - class Event_queue_element_for_exec { public: @@ -54,6 +52,24 @@ protected: MEM_ROOT mem_root; public: + /* + ENABLED = feature can function normally (is turned on) + SLAVESIDE_DISABLED = feature is turned off on slave + DISABLED = feature is turned off + */ + enum enum_status + { + ENABLED = 1, + DISABLED, + SLAVESIDE_DISABLED + }; + + enum enum_on_completion + { + ON_COMPLETION_DROP = 1, + ON_COMPLETION_PRESERVE + }; + LEX_STRING dbname; LEX_STRING name; LEX_STRING definer;// combination of user and host @@ -83,20 +99,9 @@ protected: bool last_executed_changed; public: - enum enum_status - { - ENABLED = 1, - DISABLED - }; - - enum enum_on_completion - { - ON_COMPLETION_DROP = 1, - ON_COMPLETION_PRESERVE - }; - - enum enum_on_completion on_completion; - enum enum_status status; + int on_completion; + int status; + longlong originator; my_time_t last_executed; my_time_t execute_at; @@ -216,19 +221,10 @@ private: class Event_parse_data : public Sql_alloc { public: - enum enum_status - { - ENABLED = 1, - DISABLED - }; - enum enum_on_completion - { - ON_COMPLETION_DROP = 1, - ON_COMPLETION_PRESERVE - }; - enum enum_on_completion on_completion; - enum enum_status status; + int on_completion; + int status; + longlong originator; /* do_not_create will be set if STARTS time is in the past and on_completion == ON_COMPLETION_DROP. @@ -298,6 +294,7 @@ private: check_if_in_the_past(THD *thd, my_time_t ltime_utc); Event_parse_data(const Event_parse_data &); /* Prevent use of these */ + void check_originator_id(THD *thd); void operator=(Event_parse_data &); }; diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc index 7232390ef69..bf535bf4cfa 100644 --- a/sql/event_db_repository.cc +++ b/sql/event_db_repository.cc @@ -94,7 +94,7 @@ const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] = }, { { C_STRING_WITH_LEN("status") }, - { C_STRING_WITH_LEN("enum('ENABLED','DISABLED')") }, + { C_STRING_WITH_LEN("enum('ENABLED','DISABLED','SLAVESIDE_DISABLED')") }, {NULL, 0} }, { @@ -120,6 +120,11 @@ const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] = { C_STRING_WITH_LEN("utf8") } }, { + { C_STRING_WITH_LEN("originator") }, + { C_STRING_WITH_LEN("int(10)") }, + {NULL, 0} + }, + { { C_STRING_WITH_LEN("time_zone") }, { C_STRING_WITH_LEN("char(64)") }, { C_STRING_WITH_LEN("latin1") } @@ -175,6 +180,9 @@ mysql_event_fill_row(THD *thd, TABLE *table, Event_parse_data *et, fields[ET_FIELD_STATUS]->store((longlong)et->status, TRUE); + fields[ET_FIELD_ORIGINATOR]->store((longlong)et->originator, TRUE); + + /* Change the SQL_MODE only if body was present in an ALTER EVENT and of course always during CREATE EVENT. @@ -564,7 +572,6 @@ Event_db_repository::create_event(THD *thd, Event_parse_data *parse_data, goto err; } - DBUG_PRINT("info", ("name: %.*s", parse_data->name.length, parse_data->name.str)); @@ -632,6 +639,8 @@ Event_db_repository::create_event(THD *thd, Event_parse_data *parse_data, if ((ret= mysql_event_fill_row(thd, table, parse_data, FALSE))) goto err; + table->field[ET_FIELD_STATUS]->store((longlong)parse_data->status, TRUE); + /* Close active transaction only if We are going to modify disk */ if (end_active_trans(thd)) goto err; diff --git a/sql/event_db_repository.h b/sql/event_db_repository.h index 4991e2d0aa2..0c2fcf9542d 100644 --- a/sql/event_db_repository.h +++ b/sql/event_db_repository.h @@ -19,7 +19,7 @@ enum enum_events_table_field { - ET_FIELD_DB = 0, + ET_FIELD_DB = 0, ET_FIELD_NAME, ET_FIELD_BODY, ET_FIELD_DEFINER, @@ -35,6 +35,7 @@ enum enum_events_table_field ET_FIELD_ON_COMPLETION, ET_FIELD_SQL_MODE, ET_FIELD_COMMENT, + ET_FIELD_ORIGINATOR, ET_FIELD_TIME_ZONE, ET_FIELD_COUNT /* a cool trick to count the number of fields :) */ }; diff --git a/sql/event_queue.cc b/sql/event_queue.cc index f64509ee25c..5ee0722964b 100644 --- a/sql/event_queue.cc +++ b/sql/event_queue.cc @@ -199,7 +199,8 @@ Event_queue::create_event(THD *thd, Event_queue_element *new_element) DBUG_PRINT("enter", ("thd: 0x%lx et=%s.%s", (long) thd, new_element->dbname.str, new_element->name.str)); - if (new_element->status == Event_queue_element::DISABLED) + if ((new_element->status == Event_queue_element::DISABLED) + || (new_element->status == Event_queue_element::SLAVESIDE_DISABLED)) delete new_element; else { @@ -235,7 +236,8 @@ Event_queue::update_event(THD *thd, LEX_STRING dbname, LEX_STRING name, DBUG_ENTER("Event_queue::update_event"); DBUG_PRINT("enter", ("thd: 0x%lx et=[%s.%s]", (long) thd, dbname.str, name.str)); - if (new_element->status == Event_queue_element::DISABLED) + if ((new_element->status == Event_queue_element::DISABLED) || + (new_element->status == Event_queue_element::SLAVESIDE_DISABLED)) { DBUG_PRINT("info", ("The event is disabled.")); /* diff --git a/sql/events.cc b/sql/events.cc index 88014ee34af..df2f1124f38 100644 --- a/sql/events.cc +++ b/sql/events.cc @@ -324,7 +324,15 @@ Events::create_event(THD *thd, Event_parse_data *parse_data, bool if_not_exists) DBUG_RETURN(TRUE); } + /* + Turn off row binlogging of this statement and use statement-based + so that all supporting tables are updated for CREATE EVENT command. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + pthread_mutex_lock(&LOCK_event_metadata); + /* On error conditions my_error() is called so no need to handle here */ if (!(ret= db_repository->create_event(thd, parse_data, if_not_exists)) && !parse_data->do_not_create) @@ -340,8 +348,16 @@ Events::create_event(THD *thd, Event_parse_data *parse_data, bool if_not_exists) DBUG_ASSERT(ret == OP_LOAD_ERROR); delete new_element; } - else + else /* Binlog the create event. */ + { event_queue->create_event(thd, new_element); + if (mysql_bin_log.is_open() && (thd->query_length > 0)) + { + thd->clear_error(); + thd->binlog_query(THD::MYSQL_QUERY_TYPE, + thd->query, thd->query_length, FALSE, FALSE); + } + } } pthread_mutex_unlock(&LOCK_event_metadata); @@ -383,7 +399,15 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) DBUG_RETURN(TRUE); } + /* + Turn off row binlogging of this statement and use statement-based + so that all supporting tables are updated for UPDATE EVENT command. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + pthread_mutex_lock(&LOCK_event_metadata); + /* On error conditions my_error() is called so no need to handle here */ if (!(ret= db_repository->update_event(thd, parse_data, new_dbname, new_name))) { @@ -398,9 +422,17 @@ Events::update_event(THD *thd, Event_parse_data *parse_data, sp_name *rename_to) DBUG_ASSERT(ret == OP_LOAD_ERROR); delete new_element; } - else + else /* Binlog the alter event. */ + { event_queue->update_event(thd, parse_data->dbname, parse_data->name, new_element); + if (mysql_bin_log.is_open() && (thd->query_length > 0)) + { + thd->clear_error(); + thd->binlog_query(THD::MYSQL_QUERY_TYPE, + thd->query, thd->query_length, FALSE, FALSE); + } + } } pthread_mutex_unlock(&LOCK_event_metadata); @@ -435,10 +467,26 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists) DBUG_RETURN(TRUE); } + /* + Turn off row binlogging of this statement and use statement-based so + that all supporting tables are updated for DROP EVENT command. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + pthread_mutex_lock(&LOCK_event_metadata); /* On error conditions my_error() is called so no need to handle here */ if (!(ret= db_repository->drop_event(thd, dbname, name, if_exists))) + { event_queue->drop_event(thd, dbname, name); + /* Binlog the drop event. */ + if (mysql_bin_log.is_open() && (thd->query_length > 0)) + { + thd->clear_error(); + thd->binlog_query(THD::MYSQL_QUERY_TYPE, + thd->query, thd->query_length, FALSE, FALSE); + } + } pthread_mutex_unlock(&LOCK_event_metadata); DBUG_RETURN(ret); } diff --git a/sql/field.cc b/sql/field.cc index 56e60d81e64..6b777c81fd3 100644 --- a/sql/field.cc +++ b/sql/field.cc @@ -8527,9 +8527,28 @@ char *Field_bit::pack(char *to, const char *from, uint max_length) { DBUG_ASSERT(max_length); uint length; - if (bit_len) + if (bit_len > 0) { - uchar bits= get_rec_bits(bit_ptr, bit_ofs, bit_len); + /* + We have the following: + + ptr Points into a field in record R1 + from Points to a field in a record R2 + bit_ptr Points to the byte (in the null bytes) that holds the + odd bits of R1 + from_bitp Points to the byte that holds the odd bits of R2 + + We have the following: + + ptr - bit_ptr = from - from_bitp + + We want to isolate 'from_bitp', so this gives: + + ptr - bit_ptr - from = - from_bitp + - ptr + bit_ptr + from = from_bitp + bit_ptr + from - ptr = from_bitp + */ + uchar bits= get_rec_bits(bit_ptr + (from - ptr), bit_ofs, bit_len); *to++= bits; } length= min(bytes_in_rec, max_length - (bit_len > 0)); @@ -8540,9 +8559,16 @@ char *Field_bit::pack(char *to, const char *from, uint max_length) const char *Field_bit::unpack(char *to, const char *from) { - if (bit_len) + if (bit_len > 0) { - set_rec_bits(*from, bit_ptr, bit_ofs, bit_len); + /* + set_rec_bits is a macro, don't put the post-increment in the + argument since that might cause strange side-effects. + + For the choice of the second argument, see the explanation for + Field_bit::pack(). + */ + set_rec_bits(*from, bit_ptr + (to - ptr), bit_ofs, bit_len); from++; } memcpy(to, from, bytes_in_rec); diff --git a/sql/field.h b/sql/field.h index a302fd6ce18..ad98d0b8089 100644 --- a/sql/field.h +++ b/sql/field.h @@ -193,9 +193,9 @@ public: */ virtual void sql_type(String &str) const =0; virtual uint size_of() const =0; // For new field - inline bool is_null(uint row_offset=0) + inline bool is_null(my_ptrdiff_t row_offset= 0) { return null_ptr ? (null_ptr[row_offset] & null_bit ? 1 : 0) : table->null_row; } - inline bool is_real_null(uint row_offset=0) + inline bool is_real_null(my_ptrdiff_t row_offset= 0) { return null_ptr ? (null_ptr[row_offset] & null_bit ? 1 : 0) : 0; } inline bool is_null_in_record(const uchar *record) { @@ -210,9 +210,9 @@ public: return 0; return test(null_ptr[offset] & null_bit); } - inline void set_null(int row_offset=0) + inline void set_null(my_ptrdiff_t row_offset= 0) { if (null_ptr) null_ptr[row_offset]|= null_bit; } - inline void set_notnull(int row_offset=0) + inline void set_notnull(my_ptrdiff_t row_offset= 0) { if (null_ptr) null_ptr[row_offset]&= (uchar) ~null_bit; } inline bool maybe_null(void) { return null_ptr != 0 || table->maybe_null; } inline bool real_maybe_null(void) { return null_ptr != 0; } @@ -1353,7 +1353,7 @@ public: int store_decimal(const my_decimal *); void get_key_image(char *buff,uint length,imagetype type); uint size_of() const { return sizeof(*this); } - int reset(void) { return !maybe_null(); } + int reset(void) { return !maybe_null() || Field_blob::reset(); } }; #endif /*HAVE_SPATIAL*/ diff --git a/sql/field_conv.cc b/sql/field_conv.cc index 88d36615668..a718a402897 100644 --- a/sql/field_conv.cc +++ b/sql/field_conv.cc @@ -173,7 +173,7 @@ set_field_to_null_with_conversions(Field *field, bool no_conversions) if (field == field->table->next_number_field) { field->table->auto_increment_field_not_null= FALSE; - return 0; // field is set in handler.cc + return 0; // field is set in fill_record() } if (field->table->in_use->count_cuted_fields == CHECK_FIELD_WARN) { diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc index 8e8aa162fff..353dd806448 100644 --- a/sql/ha_ndbcluster.cc +++ b/sql/ha_ndbcluster.cc @@ -3873,7 +3873,8 @@ int ha_ndbcluster::extra(enum ha_extra_function operation) break; case HA_EXTRA_WRITE_CAN_REPLACE: DBUG_PRINT("info", ("HA_EXTRA_WRITE_CAN_REPLACE")); - if (!m_has_unique_index) + if (!m_has_unique_index || + current_thd->slave_thread) /* always set if slave, quick fix for bug 27378 */ { DBUG_PRINT("info", ("Turning ON use of write instead of insert")); m_use_write= TRUE; diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc index 8b51ac68e66..4dc75be79e9 100644 --- a/sql/ha_ndbcluster_binlog.cc +++ b/sql/ha_ndbcluster_binlog.cc @@ -500,7 +500,7 @@ static int ndbcluster_reset_logs(THD *thd) static int ndbcluster_binlog_index_purge_file(THD *thd, const char *file) { - if (!ndb_binlog_running) + if (!ndb_binlog_running || thd->slave_thread) return 0; DBUG_ENTER("ndbcluster_binlog_index_purge_file"); diff --git a/sql/handler.cc b/sql/handler.cc index 0ecfce9349e..617bf9ee378 100644 --- a/sql/handler.cc +++ b/sql/handler.cc @@ -1753,7 +1753,6 @@ int handler::update_auto_increment() bool append= FALSE; THD *thd= table->in_use; struct system_variables *variables= &thd->variables; - bool auto_increment_field_not_null; DBUG_ENTER("handler::update_auto_increment"); /* @@ -1761,11 +1760,9 @@ int handler::update_auto_increment() than the interval, but not smaller. */ DBUG_ASSERT(next_insert_id >= auto_inc_interval_for_cur_row.minimum()); - auto_increment_field_not_null= table->auto_increment_field_not_null; - table->auto_increment_field_not_null= FALSE; // to reset for next row if ((nr= table->next_number_field->val_int()) != 0 || - auto_increment_field_not_null && + table->auto_increment_field_not_null && thd->variables.sql_mode & MODE_NO_AUTO_VALUE_ON_ZERO) { /* diff --git a/sql/item.cc b/sql/item.cc index 3e3ef2a0bbe..3f4a121b6af 100644 --- a/sql/item.cc +++ b/sql/item.cc @@ -1106,7 +1106,7 @@ bool Item_splocal::set_value(THD *thd, sp_rcontext *ctx, Item **it) Item_case_expr methods *****************************************************************************/ -Item_case_expr::Item_case_expr(int case_expr_id) +Item_case_expr::Item_case_expr(uint case_expr_id) :Item_sp_variable( C_STRING_WITH_LEN("case_expr")), m_case_expr_id(case_expr_id) { @@ -1143,6 +1143,8 @@ Item_case_expr::this_item_addr(THD *thd, Item **) void Item_case_expr::print(String *str) { + if (str->reserve(MAX_INT_WIDTH + sizeof("case_expr@"))) + return; /* purecov: inspected */ VOID(str->append(STRING_WITH_LEN("case_expr@"))); str->qs_append(m_case_expr_id); } @@ -1307,15 +1309,18 @@ void Item::split_sum_func2(THD *thd, Item **ref_pointer_array, Exception is Item_direct_view_ref which we need to convert to Item_ref to allow fields from view being stored in tmp table. */ + Item_aggregate_ref *item_ref; uint el= fields.elements; - Item *new_item, *real_itm= real_item(); + Item *real_itm= real_item(); ref_pointer_array[el]= real_itm; - if (!(new_item= new Item_aggregate_ref(&thd->lex->current_select->context, + if (!(item_ref= new Item_aggregate_ref(&thd->lex->current_select->context, ref_pointer_array + el, 0, name))) return; // fatal_error is set + if (type() == SUM_FUNC_ITEM) + item_ref->depended_from= ((Item_sum *) this)->depended_from(); fields.push_front(real_itm); - thd->change_item_tree(ref, new_item); + thd->change_item_tree(ref, item_ref); } } @@ -4377,13 +4382,16 @@ Field *Item::tmp_table_field_from_field_type(TABLE *table, bool fixed_length) case MYSQL_TYPE_MEDIUM_BLOB: case MYSQL_TYPE_LONG_BLOB: case MYSQL_TYPE_BLOB: - case MYSQL_TYPE_GEOMETRY: if (this->type() == Item::TYPE_HOLDER) field= new Field_blob(max_length, maybe_null, name, collation.collation, 1); else field= new Field_blob(max_length, maybe_null, name, collation.collation); break; // Blob handled outside of case + case MYSQL_TYPE_GEOMETRY: + return new Field_geom(max_length, maybe_null, name, table->s, + (Field::geometry_type) + ((Item_geometry_func *)this)->get_geometry_type()); } if (field) field->init(table); diff --git a/sql/item.h b/sql/item.h index ee178a1e379..6be7cccd84f 100644 --- a/sql/item.h +++ b/sql/item.h @@ -1116,7 +1116,7 @@ inline Item_result Item_splocal::result_type() const class Item_case_expr :public Item_sp_variable { public: - Item_case_expr(int case_expr_id); + Item_case_expr(uint case_expr_id); public: Item *this_item(); @@ -1135,7 +1135,7 @@ public: void print(String *str); private: - int m_case_expr_id; + uint m_case_expr_id; }; /***************************************************************************** diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc index a011a2884c8..e6f951bfc7b 100644 --- a/sql/item_strfunc.cc +++ b/sql/item_strfunc.cc @@ -1813,7 +1813,8 @@ void Item_func_soundex::fix_length_and_dec() { collation.set(args[0]->collation); max_length=args[0]->max_length; - set_if_bigger(max_length,4); + set_if_bigger(max_length, 4 * collation.collation->mbminlen); + tmp_value.set_charset(collation.collation); } @@ -1823,14 +1824,15 @@ void Item_func_soundex::fix_length_and_dec() else return 0 */ -static char soundex_toupper(char ch) +static int soundex_toupper(int ch) { return (ch >= 'a' && ch <= 'z') ? ch - 'a' + 'A' : ch; } -static char get_scode(char *ptr) + +static char get_scode(int wc) { - uchar ch= soundex_toupper(*ptr); + int ch= soundex_toupper(wc); if (ch < 'A' || ch > 'Z') { // Thread extended alfa (country spec) @@ -1840,46 +1842,121 @@ static char get_scode(char *ptr) } +static bool my_uni_isalpha(int wc) +{ + /* + Return true for all Basic Latin letters: a..z A..Z. + Return true for all Unicode characters with code higher than U+00C0: + - characters between 'z' and U+00C0 are controls and punctuations. + - "U+00C0 LATIN CAPITAL LETTER A WITH GRAVE" is the first letter after 'z'. + */ + return (wc >= 'a' && wc <= 'z') || + (wc >= 'A' && wc <= 'Z') || + (wc >= 0xC0); +} + + String *Item_func_soundex::val_str(String *str) { DBUG_ASSERT(fixed == 1); String *res =args[0]->val_str(str); char last_ch,ch; CHARSET_INFO *cs= collation.collation; + my_wc_t wc; + uint nchars; + int rc; - if ((null_value=args[0]->null_value)) + if ((null_value= args[0]->null_value)) return 0; /* purecov: inspected */ - if (tmp_value.alloc(max(res->length(),4))) + if (tmp_value.alloc(max(res->length(), 4 * cs->mbminlen))) return str; /* purecov: inspected */ char *to= (char *) tmp_value.ptr(); - char *from= (char *) res->ptr(), *end=from+res->length(); - tmp_value.set_charset(cs); + char *to_end= to + tmp_value.alloced_length(); + char *from= (char *) res->ptr(), *end= from + res->length(); - while (from != end && !my_isalpha(cs,*from)) // Skip pre-space - from++; /* purecov: inspected */ - if (from == end) - return &my_empty_string; // No alpha characters. - *to++ = soundex_toupper(*from); // Copy first letter - last_ch = get_scode(from); // code of the first letter - // for the first 'double-letter check. - // Loop on input letters until - // end of input (null) or output - // letter code count = 3 - for (from++ ; from < end ; from++) - { - if (!my_isalpha(cs,*from)) - continue; - ch=get_scode(from); + for ( ; ; ) /* Skip pre-space */ + { + if ((rc= cs->cset->mb_wc(cs, &wc, (uchar*) from, (uchar*) end)) <= 0) + return &my_empty_string; /* EOL or invalid byte sequence */ + + if (rc == 1 && cs->ctype) + { + /* Single byte letter found */ + if (my_isalpha(cs, *from)) + { + last_ch= get_scode(*from); // Code of the first letter + *to++= soundex_toupper(*from++); // Copy first letter + break; + } + from++; + } + else + { + from+= rc; + if (my_uni_isalpha(wc)) + { + /* Multibyte letter found */ + wc= soundex_toupper(wc); + last_ch= get_scode(wc); // Code of the first letter + if ((rc= cs->cset->wc_mb(cs, wc, (uchar*) to, (uchar*) to_end)) <= 0) + { + /* Extra safety - should not really happen */ + DBUG_ASSERT(false); + return &my_empty_string; + } + to+= rc; + break; + } + } + } + + /* + last_ch is now set to the first 'double-letter' check. + loop on input letters until end of input + */ + for (nchars= 1 ; ; ) + { + if ((rc= cs->cset->mb_wc(cs, &wc, (uchar*) from, (uchar*) end)) <= 0) + break; /* EOL or invalid byte sequence */ + + if (rc == 1 && cs->ctype) + { + if (!my_isalpha(cs, *from++)) + continue; + } + else + { + from+= rc; + if (!my_uni_isalpha(wc)) + continue; + } + + ch= get_scode(wc); if ((ch != '0') && (ch != last_ch)) // if not skipped or double { - *to++ = ch; // letter, copy to output - last_ch = ch; // save code of last input letter - } // for next double-letter check + // letter, copy to output + if ((rc= cs->cset->wc_mb(cs, (my_wc_t) ch, + (uchar*) to, (uchar*) to_end)) <= 0) + { + // Extra safety - should not really happen + DBUG_ASSERT(false); + break; + } + to+= rc; + nchars++; + last_ch= ch; // save code of last input letter + } // for next double-letter check } - for (end=(char*) tmp_value.ptr()+4 ; to < end ; to++) - *to = '0'; - *to=0; // end string + + /* Pad up to 4 characters with DIGIT ZERO, if the string is shorter */ + if (nchars < 4) + { + uint nbytes= (4 - nchars) * cs->mbminlen; + cs->cset->fill(cs, to, nbytes, '0'); + to+= nbytes; + } + tmp_value.length((uint) (to-tmp_value.ptr())); return &tmp_value; } diff --git a/sql/item_sum.cc b/sql/item_sum.cc index cd67f83af17..7c3762d4785 100644 --- a/sql/item_sum.cc +++ b/sql/item_sum.cc @@ -61,9 +61,9 @@ bool Item_sum::init_sum_func_check(THD *thd) /* Save a pointer to object to be used in items for nested set functions */ thd->lex->in_sum_func= this; nest_level= thd->lex->current_select->nest_level; - nest_level_tables_count= thd->lex->current_select->join->tables; ref_by= 0; aggr_level= -1; + aggr_sel= NULL; max_arg_level= -1; max_sum_func_level= -1; return FALSE; @@ -149,9 +149,14 @@ bool Item_sum::check_sum_func(THD *thd, Item **ref) if (register_sum_func(thd, ref)) return TRUE; invalid= aggr_level < 0 && !(allow_sum_func & (1 << nest_level)); + if (!invalid && thd->variables.sql_mode & MODE_ANSI) + invalid= aggr_level < 0 && max_arg_level < nest_level; } if (!invalid && aggr_level < 0) + { aggr_level= nest_level; + aggr_sel= thd->lex->current_select; + } /* By this moment we either found a subquery where the set function is to be aggregated and assigned a value that is >= 0 to aggr_level, @@ -161,8 +166,9 @@ bool Item_sum::check_sum_func(THD *thd, Item **ref) Additionally we have to check whether possible nested set functions are acceptable here: they are not, if the level of aggregation of some of them is less than aggr_level. - */ - invalid= aggr_level <= max_sum_func_level; + */ + if (!invalid) + invalid= aggr_level <= max_sum_func_level; if (invalid) { my_message(ER_INVALID_GROUP_FUNC_USE, ER(ER_INVALID_GROUP_FUNC_USE), @@ -212,7 +218,6 @@ bool Item_sum::check_sum_func(THD *thd, Item **ref) bool Item_sum::register_sum_func(THD *thd, Item **ref) { SELECT_LEX *sl; - SELECT_LEX *aggr_sl= NULL; nesting_map allow_sum_func= thd->lex->allow_sum_func; for (sl= thd->lex->current_select->master_unit()->outer_select() ; sl && sl->nest_level > max_arg_level; @@ -222,7 +227,7 @@ bool Item_sum::register_sum_func(THD *thd, Item **ref) { /* Found the most nested subquery where the function can be aggregated */ aggr_level= sl->nest_level; - aggr_sl= sl; + aggr_sel= sl; } } if (sl && (allow_sum_func & (1 << sl->nest_level))) @@ -233,21 +238,22 @@ bool Item_sum::register_sum_func(THD *thd, Item **ref) The set function will be aggregated in this subquery. */ aggr_level= sl->nest_level; - aggr_sl= sl; + aggr_sel= sl; + } if (aggr_level >= 0) { ref_by= ref; - /* Add the object to the list of registered objects assigned to aggr_sl */ - if (!aggr_sl->inner_sum_func_list) + /* Add the object to the list of registered objects assigned to aggr_sel */ + if (!aggr_sel->inner_sum_func_list) next= this; else { - next= aggr_sl->inner_sum_func_list->next; - aggr_sl->inner_sum_func_list->next= this; + next= aggr_sel->inner_sum_func_list->next; + aggr_sel->inner_sum_func_list->next= this; } - aggr_sl->inner_sum_func_list= this; - aggr_sl->with_sum_func= 1; + aggr_sel->inner_sum_func_list= this; + aggr_sel->with_sum_func= 1; /* Mark Item_subselect(s) as containing aggregate function all the way up @@ -265,11 +271,11 @@ bool Item_sum::register_sum_func(THD *thd, Item **ref) has aggregate functions directly referenced (i.e. not through a sub-select). */ for (sl= thd->lex->current_select; - sl && sl != aggr_sl && sl->master_unit()->item; + sl && sl != aggr_sel && sl->master_unit()->item; sl= sl->master_unit()->outer_select() ) sl->master_unit()->item->with_sum_func= 1; } - thd->lex->current_select->mark_as_dependent(aggr_sl); + thd->lex->current_select->mark_as_dependent(aggr_sel); return FALSE; } @@ -299,10 +305,10 @@ Item_sum::Item_sum(List<Item> &list) :arg_count(list.elements), Item_sum::Item_sum(THD *thd, Item_sum *item): Item_result_field(thd, item), arg_count(item->arg_count), + aggr_sel(item->aggr_sel), nest_level(item->nest_level), aggr_level(item->aggr_level), quick_group(item->quick_group), used_tables_cache(item->used_tables_cache), - forced_const(item->forced_const), - nest_level_tables_count(item->nest_level_tables_count) + forced_const(item->forced_const) { if (arg_count <= 2) args=tmp_args; @@ -413,7 +419,7 @@ Field *Item_sum::create_tmp_field(bool group, TABLE *table, break; case STRING_RESULT: if (max_length/collation.collation->mbmaxlen <= 255 || - max_length/collation.collation->mbmaxlen >=UINT_MAX16 || + convert_blob_length >=UINT_MAX16 || !convert_blob_length) return make_string_field(table); field= new Field_varstring(convert_blob_length, maybe_null, @@ -449,8 +455,7 @@ void Item_sum::update_used_tables () used_tables_cache&= PSEUDO_TABLE_BITS; /* the aggregate function is aggregated into its local context */ - if (aggr_level == nest_level) - used_tables_cache |= (1 << nest_level_tables_count) - 1; + used_tables_cache |= (1 << aggr_sel->join->tables) - 1; } } diff --git a/sql/item_sum.h b/sql/item_sum.h index 5756402c1b1..5cf4f93af0e 100644 --- a/sql/item_sum.h +++ b/sql/item_sum.h @@ -233,6 +233,7 @@ public: Item_sum *next; /* next in the circular chain of registered objects */ uint arg_count; Item_sum *in_sum_func; /* embedding set function if any */ + st_select_lex * aggr_sel; /* select where the function is aggregated */ int8 nest_level; /* number of the nesting level of the set function */ int8 aggr_level; /* nesting level of the aggregating subquery */ int8 max_arg_level; /* max level of unbound column references */ @@ -242,7 +243,6 @@ public: protected: table_map used_tables_cache; bool forced_const; - byte nest_level_tables_count; public: @@ -365,6 +365,8 @@ public: bool init_sum_func_check(THD *thd); bool check_sum_func(THD *thd, Item **ref); bool register_sum_func(THD *thd, Item **ref); + st_select_lex *depended_from() + { return (nest_level == aggr_level ? 0 : aggr_sel); } }; diff --git a/sql/lock.cc b/sql/lock.cc index 2373b62eb0a..4427e57a938 100644 --- a/sql/lock.cc +++ b/sql/lock.cc @@ -582,7 +582,7 @@ TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle, goto end; /* A temporary table does not have locks. */ - if (table->s->tmp_table == TMP_TABLE) + if (table->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE) goto end; /* Get command lock or LOCK TABLES lock. Maybe empty for INSERT DELAYED. */ @@ -607,7 +607,7 @@ TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle, if (haystack->placeholder()) continue; table2= haystack->table; - if (table2->s->tmp_table == TMP_TABLE) + if (table2->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE) continue; /* All tables in list must be in lock. */ @@ -695,7 +695,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count, uint system_count= 0; for (i=tables=lock_count=0 ; i < count ; i++) { - if (table_ptr[i]->s->tmp_table != TMP_TABLE) + if (table_ptr[i]->s->tmp_table != NON_TRANSACTIONAL_TMP_TABLE) { tables+=table_ptr[i]->file->lock_count(); lock_count++; @@ -737,7 +737,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count, TABLE *table; enum thr_lock_type lock_type; - if ((table=table_ptr[i])->s->tmp_table == TMP_TABLE) + if ((table=table_ptr[i])->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE) continue; lock_type= table->reginfo.lock_type; if (lock_type >= TL_WRITE_ALLOW_WRITE) diff --git a/sql/log.cc b/sql/log.cc index 4a51c52056d..fe3a4f7df5e 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -2499,7 +2499,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name, /* Set 'created' to 0, so that in next relay logs this event does not trigger cleaning actions on the slave in - Format_description_log_event::exec_event(). + Format_description_log_event::apply_event_impl(). */ description_event_for_queue->created= 0; /* Don't set log_pos in event header */ @@ -3206,8 +3206,10 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock) { tc_log_page_waits++; pthread_mutex_lock(&LOCK_prep_xids); - while (prepared_xids) + while (prepared_xids) { + DBUG_PRINT("info", ("prepared_xids=%lu", prepared_xids)); pthread_cond_wait(&COND_prep_xids, &LOCK_prep_xids); + } pthread_mutex_unlock(&LOCK_prep_xids); } @@ -5061,8 +5063,10 @@ void TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid) { pthread_mutex_lock(&LOCK_prep_xids); DBUG_ASSERT(prepared_xids > 0); - if (--prepared_xids == 0) + if (--prepared_xids == 0) { + DBUG_PRINT("info", ("prepared_xids=%lu", prepared_xids)); pthread_cond_signal(&COND_prep_xids); + } pthread_mutex_unlock(&LOCK_prep_xids); rotate_and_purge(0); // as ::write() did not rotate } diff --git a/sql/log_event.cc b/sql/log_event.cc index 8714c60f2c0..e3c94b5e1c9 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -75,8 +75,7 @@ public: ~Write_on_release_cache() { - if (!my_b_copy_to_file(m_cache, m_file)) - reinit_io_cache(m_cache, WRITE_CACHE, 0L, FALSE, TRUE); + copy_event_cache_to_file_and_reinit(m_cache, m_file); if (m_flags | FLUSH_F) fflush(m_file); } @@ -88,9 +87,10 @@ public: operator&() DESCRIPTION - Function to return a pointer to the internal, so that the object - can be treated as a IO_CACHE and used with the my_b_* IO_CACHE - functions + + Function to return a pointer to the internal cache, so that the + object can be treated as a IO_CACHE and used with the my_b_* + IO_CACHE functions RETURN VALUE A pointer to the internal IO_CACHE. @@ -305,7 +305,7 @@ static bool write_str(IO_CACHE *file, char *str, uint length) read_str() */ -static inline int read_str(char **buf, char *buf_end, char **str, +static inline int read_str(const char **buf, const char *buf_end, const char **str, uint8 *len) { if (*buf + ((uint) (uchar) **buf) >= buf_end) @@ -420,6 +420,7 @@ const char* Log_event::get_type_str() case DELETE_ROWS_EVENT: return "Delete_rows"; case BEGIN_LOAD_QUERY_EVENT: return "Begin_load_query"; case EXECUTE_LOAD_QUERY_EVENT: return "Execute_load_query"; + case INCIDENT_EVENT: return "Incident"; default: return "Unknown"; /* impossible */ } } @@ -531,25 +532,19 @@ Log_event::Log_event(const char* buf, #ifndef MYSQL_CLIENT #ifdef HAVE_REPLICATION -/* - Log_event::exec_event() -*/ - -int Log_event::exec_event(struct st_relay_log_info* rli) +int Log_event::do_update_pos(RELAY_LOG_INFO *rli) { - DBUG_ENTER("Log_event::exec_event"); - /* - rli is null when (as far as I (Guilhem) know) - the caller is - Load_log_event::exec_event *and* that one is called from - Execute_load_log_event::exec_event. - In this case, we don't do anything here ; - Execute_load_log_event::exec_event will call Log_event::exec_event - again later with the proper rli. - Strictly speaking, if we were sure that rli is null - only in the case discussed above, 'if (rli)' is useless here. - But as we are not 100% sure, keep it for now. + rli is null when (as far as I (Guilhem) know) the caller is + Load_log_event::do_apply_event *and* that one is called from + Execute_load_log_event::do_apply_event. In this case, we don't + do anything here ; Execute_load_log_event::do_apply_event will + call Log_event::do_apply_event again later with the proper rli. + Strictly speaking, if we were sure that rli is null only in the + case discussed above, 'if (rli)' is useless here. But as we are + not 100% sure, keep it for now. + + Matz: I don't think we will need this check with this refactoring. */ if (rli) { @@ -584,18 +579,37 @@ int Log_event::exec_event(struct st_relay_log_info* rli) { rli->inc_group_relay_log_pos(log_pos); flush_relay_log_info(rli); - /* - Note that Rotate_log_event::exec_event() does not call this - function, so there is no chance that a fake rotate event resets - last_master_timestamp. - Note that we update without mutex (probably ok - except in some very - rare cases, only consequence is that value may take some time to - display in Seconds_Behind_Master - not critical). + /* + Note that Rotate_log_event::do_apply_event() does not call + this function, so there is no chance that a fake rotate event + resets last_master_timestamp. Note that we update without + mutex (probably ok - except in some very rare cases, only + consequence is that value may take some time to display in + Seconds_Behind_Master - not critical). */ rli->last_master_timestamp= when; } } - DBUG_RETURN(0); + + return 0; // Cannot fail currently +} + + +Log_event::enum_skip_reason +Log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + DBUG_PRINT("info", ("ev->server_id=%lu, ::server_id=%lu," + " rli->replicate_same_server_id=%d," + " rli->slave_skip_counter=%d", + (ulong) server_id, (ulong) ::server_id, + rli->replicate_same_server_id, + rli->slave_skip_counter)); + if (server_id == ::server_id && !rli->replicate_same_server_id) + return EVENT_SKIP_IGNORE; + else if (rli->slave_skip_counter > 0) + return EVENT_SKIP_COUNT; + else + return EVENT_SKIP_NOT; } @@ -743,7 +757,7 @@ int Log_event::read_log_event(IO_CACHE* file, String* packet, ulong data_len; int result=0; char buf[LOG_EVENT_MINIMAL_HEADER_LEN]; - DBUG_ENTER("read_log_event"); + DBUG_ENTER("Log_event::read_log_event"); if (log_lock) pthread_mutex_lock(log_lock); @@ -818,7 +832,7 @@ Log_event* Log_event::read_log_event(IO_CACHE* file, const Format_description_log_event *description_event) #endif { - DBUG_ENTER("Log_event::read_log_event(IO_CACHE *, Format_description_log_event *"); + DBUG_ENTER("Log_event::read_log_event"); DBUG_ASSERT(description_event != 0); char head[LOG_EVENT_MINIMAL_HEADER_LEN]; /* @@ -998,10 +1012,14 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len, ev = new Begin_load_query_log_event(buf, event_len, description_event); break; case EXECUTE_LOAD_QUERY_EVENT: - ev = new Execute_load_query_log_event(buf, event_len, description_event); + ev= new Execute_load_query_log_event(buf, event_len, description_event); + break; + case INCIDENT_EVENT: + ev = new Incident_log_event(buf, event_len, description_event); break; default: - DBUG_PRINT("error",("Unknown evernt code: %d",(int) buf[EVENT_TYPE_OFFSET])); + DBUG_PRINT("error",("Unknown event code: %d", + (int) buf[EVENT_TYPE_OFFSET])); ev= NULL; break; } @@ -1888,27 +1906,28 @@ void Query_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) /* - Query_log_event::exec_event() + Query_log_event::do_apply_event() */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Query_log_event::exec_event(struct st_relay_log_info* rli) +int Query_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { - return exec_event(rli, query, q_len); + return do_apply_event(rli, query, q_len); } -int Query_log_event::exec_event(struct st_relay_log_info* rli, - const char *query_arg, uint32 q_len_arg) +int Query_log_event::do_apply_event(RELAY_LOG_INFO const *rli, + const char *query_arg, uint32 q_len_arg) { LEX_STRING new_db; int expected_error,actual_error= 0; /* - Colleagues: please never free(thd->catalog) in MySQL. This would lead to - bugs as here thd->catalog is a part of an alloced block, not an entire - alloced block (see Query_log_event::exec_event()). Same for thd->db. - Thank you. + Colleagues: please never free(thd->catalog) in MySQL. This would + lead to bugs as here thd->catalog is a part of an alloced block, + not an entire alloced block (see + Query_log_event::do_apply_event()). Same for thd->db. Thank + you. */ thd->catalog= catalog_len ? (char *) catalog : (char *)""; new_db.length= db_len; @@ -1927,11 +1946,11 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli, END of the current log event (COMMIT). We save it in rli so that InnoDB can access it. */ - rli->future_group_master_log_pos= log_pos; + const_cast<RELAY_LOG_INFO*>(rli)->future_group_master_log_pos= log_pos; DBUG_PRINT("info", ("log_pos: %lu", (ulong) log_pos)); - clear_all_errors(thd, rli); - rli->clear_tables_to_lock(); + clear_all_errors(thd, const_cast<RELAY_LOG_INFO*>(rli)); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); /* Note: We do not need to execute reset_one_shot_variables() if this @@ -1940,8 +1959,8 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli, its companion query. If the SET is ignored because of db_ok(), the companion query will also be ignored, and if the companion query is ignored in the db_ok() test of - ::exec_event(), then the companion SET also have so we - don't need to reset_one_shot_variables(). + ::do_apply_event(), then the companion SET also have so + we don't need to reset_one_shot_variables(). */ if (rpl_filter->db_ok(thd->db)) { @@ -2056,7 +2075,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli, to check/fix it. */ if (mysql_test_parse_for_slave(thd, thd->query, thd->query_length)) - clear_all_errors(thd, rli); /* Can ignore query */ + clear_all_errors(thd, const_cast<RELAY_LOG_INFO*>(rli)); /* Can ignore query */ else { slave_print_msg(ERROR_LEVEL, rli, expected_error, @@ -2107,7 +2126,7 @@ Default database: '%s'. Query: '%s'", ignored_error_code(actual_error)) { DBUG_PRINT("info",("error ignored")); - clear_all_errors(thd, rli); + clear_all_errors(thd, const_cast<RELAY_LOG_INFO*>(rli)); } /* Other cases: mostly we expected no error and get one. @@ -2174,16 +2193,26 @@ end: thd->first_successful_insert_id_in_prev_stmt= 0; thd->stmt_depends_on_first_successful_insert_id_in_prev_stmt= 0; free_root(thd->mem_root,MYF(MY_KEEP_PREALLOC)); + return thd->query_error; +} + +int Query_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ /* - If there was an error we stop. Otherwise we increment positions. Note that - we will not increment group* positions if we are just after a SET - ONE_SHOT, because SET ONE_SHOT should not be separated from its following - updating query. + Note that we will not increment group* positions if we are just + after a SET ONE_SHOT, because SET ONE_SHOT should not be separated + from its following updating query. */ - return (thd->query_error ? thd->query_error : - (thd->one_shot_set ? (rli->inc_event_relay_log_pos(),0) : - Log_event::exec_event(rli))); + if (thd->one_shot_set) + { + rli->inc_event_relay_log_pos(); + return 0; + } + else + return Log_event::do_update_pos(rli); } + + #endif @@ -2312,7 +2341,7 @@ bool Start_log_event_v3::write(IO_CACHE* file) /* - Start_log_event_v3::exec_event() + Start_log_event_v3::do_apply_event() The master started @@ -2331,9 +2360,9 @@ bool Start_log_event_v3::write(IO_CACHE* file) */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Start_log_event_v3::exec_event(struct st_relay_log_info* rli) +int Start_log_event_v3::do_apply_event(RELAY_LOG_INFO const *rli) { - DBUG_ENTER("Start_log_event_v3::exec_event"); + DBUG_ENTER("Start_log_event_v3::do_apply_event"); switch (binlog_version) { case 3: @@ -2375,7 +2404,7 @@ int Start_log_event_v3::exec_event(struct st_relay_log_info* rli) /* this case is impossible */ DBUG_RETURN(1); } - DBUG_RETURN(Log_event::exec_event(rli)); + DBUG_RETURN(0); } #endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */ @@ -2456,6 +2485,7 @@ Format_description_log_event(uint8 binlog_ver, const char* server_ver) post_header_len[DELETE_ROWS_EVENT-1]= 6;); post_header_len[BEGIN_LOAD_QUERY_EVENT-1]= post_header_len[APPEND_BLOCK_EVENT-1]; post_header_len[EXECUTE_LOAD_QUERY_EVENT-1]= EXECUTE_LOAD_QUERY_HEADER_LEN; + post_header_len[INCIDENT_EVENT-1]= INCIDENT_HEADER_LEN; } break; @@ -2566,24 +2596,10 @@ bool Format_description_log_event::write(IO_CACHE* file) } #endif -/* - SYNOPSIS - Format_description_log_event::exec_event() - - IMPLEMENTATION - Save the information which describes the binlog's format, to be able to - read all coming events. - Call Start_log_event_v3::exec_event(). -*/ - #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Format_description_log_event::exec_event(struct st_relay_log_info* rli) +int Format_description_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { - DBUG_ENTER("Format_description_log_event::exec_event"); - - /* save the information describing this binlog */ - delete rli->relay_log.description_event_for_exec; - rli->relay_log.description_event_for_exec= this; + DBUG_ENTER("Format_description_log_event::do_apply_event"); #ifdef USING_TRANSACTIONS /* @@ -2605,14 +2621,36 @@ int Format_description_log_event::exec_event(struct st_relay_log_info* rli) "or ROLLBACK in relay log). A probable cause is that " "the master died while writing the transaction to " "its binary log, thus rolled back too."); - rli->cleanup_context(thd, 1); + const_cast<RELAY_LOG_INFO*>(rli)->cleanup_context(thd, 1); } #endif /* - If this event comes from ourselves, there is no cleaning task to perform, - we don't call Start_log_event_v3::exec_event() (this was just to update the - log's description event). + If this event comes from ourselves, there is no cleaning task to + perform, we don't call Start_log_event_v3::do_apply_event() + (this was just to update the log's description event). */ + if (server_id != (uint32) ::server_id) + { + /* + If the event was not requested by the slave i.e. the master sent + it while the slave asked for a position >4, the event will make + rli->group_master_log_pos advance. Say that the slave asked for + position 1000, and the Format_desc event's end is 96. Then in + the beginning of replication rli->group_master_log_pos will be + 0, then 96, then jump to first really asked event (which is + >96). So this is ok. + */ + DBUG_RETURN(Start_log_event_v3::do_apply_event(rli)); + } + DBUG_RETURN(0); +} + +int Format_description_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ + /* save the information describing this binlog */ + delete rli->relay_log.description_event_for_exec; + rli->relay_log.description_event_for_exec= this; + if (server_id == (uint32) ::server_id) { /* @@ -2629,19 +2667,20 @@ int Format_description_log_event::exec_event(struct st_relay_log_info* rli) the Intvar_log_event respectively. */ rli->inc_event_relay_log_pos(); - DBUG_RETURN(0); + return 0; + } + else + { + return Log_event::do_update_pos(rli); } +} - /* - If the event was not requested by the slave i.e. the master sent it while - the slave asked for a position >4, the event will make - rli->group_master_log_pos advance. Say that the slave asked for position - 1000, and the Format_desc event's end is 96. Then in the beginning of - replication rli->group_master_log_pos will be 0, then 96, then jump to - first really asked event (which is >96). So this is ok. - */ - DBUG_RETURN(Start_log_event_v3::exec_event(rli)); +Log_event::enum_skip_reason +Format_description_log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + return Log_event::EVENT_SKIP_NOT; } + #endif @@ -3155,30 +3194,32 @@ void Load_log_event::set_fields(const char* affected_db, Does the data loading job when executing a LOAD DATA on the slave SYNOPSIS - Load_log_event::exec_event - net - rli - use_rli_only_for_errors - if set to 1, rli is provided to - Load_log_event::exec_event only for this - function to have RPL_LOG_NAME and - rli->last_slave_error, both being used by - error reports. rli's position advancing - is skipped (done by the caller which is - Execute_load_log_event::exec_event). - - if set to 0, rli is provided for full use, - i.e. for error reports and position - advancing. + Load_log_event::do_apply_event + net + rli + use_rli_only_for_errors - if set to 1, rli is provided to + Load_log_event::do_apply_event + only for this function to have + RPL_LOG_NAME and + rli->last_slave_error, both being + used by error reports. rli's + position advancing is skipped (done + by the caller which is + Execute_load_log_event::do_apply_event). + - if set to 0, rli is provided for + full use, i.e. for error reports and + position advancing. DESCRIPTION Does the data loading job when executing a LOAD DATA on the slave - + RETURN VALUE - 0 Success + 0 Success 1 Failure */ -int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, - bool use_rli_only_for_errors) +int Load_log_event::do_apply_event(NET* net, RELAY_LOG_INFO const *rli, + bool use_rli_only_for_errors) { LEX_STRING new_db; new_db.length= db_len; @@ -3187,9 +3228,9 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, DBUG_ASSERT(thd->query == 0); thd->query_length= 0; // Should not be needed thd->query_error= 0; - clear_all_errors(thd, rli); + clear_all_errors(thd, const_cast<RELAY_LOG_INFO*>(rli)); - /* see Query_log_event::exec_event() and BUG#13360 */ + /* see Query_log_event::do_apply_event() and BUG#13360 */ DBUG_ASSERT(!rli->m_table_map.count()); /* Usually mysql_init_query() is called by mysql_parse(), but we need it here @@ -3198,22 +3239,26 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, mysql_init_query(thd, 0, 0); if (!use_rli_only_for_errors) { - /* Saved for InnoDB, see comment in Query_log_event::exec_event() */ - rli->future_group_master_log_pos= log_pos; + /* + Saved for InnoDB, see comment in + Query_log_event::do_apply_event() + */ + const_cast<RELAY_LOG_INFO*>(rli)->future_group_master_log_pos= log_pos; DBUG_PRINT("info", ("log_pos: %lu", (ulong) log_pos)); } /* - We test replicate_*_db rules. Note that we have already prepared the file - to load, even if we are going to ignore and delete it now. So it is - possible that we did a lot of disk writes for nothing. In other words, a - big LOAD DATA INFILE on the master will still consume a lot of space on - the slave (space in the relay log + space of temp files: twice the space - of the file to load...) even if it will finally be ignored. - TODO: fix this; this can be done by testing rules in - Create_file_log_event::exec_event() and then discarding Append_block and - al. Another way is do the filtering in the I/O thread (more efficient: no - disk writes at all). + We test replicate_*_db rules. Note that we have already prepared + the file to load, even if we are going to ignore and delete it + now. So it is possible that we did a lot of disk writes for + nothing. In other words, a big LOAD DATA INFILE on the master will + still consume a lot of space on the slave (space in the relay log + + space of temp files: twice the space of the file to load...) + even if it will finally be ignored. TODO: fix this; this can be + done by testing rules in Create_file_log_event::do_apply_event() + and then discarding Append_block and al. Another way is do the + filtering in the I/O thread (more efficient: no disk writes at + all). Note: We do not need to execute reset_one_shot_variables() if this @@ -3222,8 +3267,8 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, its companion query. If the SET is ignored because of db_ok(), the companion query will also be ignored, and if the companion query is ignored in the db_ok() test of - ::exec_event(), then the companion SET also have so we - don't need to reset_one_shot_variables(). + ::do_apply_event(), then the companion SET also have so + we don't need to reset_one_shot_variables(). */ if (rpl_filter->db_ok(thd->db)) { @@ -3419,7 +3464,7 @@ Fatal error running LOAD DATA INFILE on table '%s'. Default database: '%s'", return 1; } - return ( use_rli_only_for_errors ? 0 : Log_event::exec_event(rli) ); + return ( use_rli_only_for_errors ? 0 : Log_event::do_apply_event(rli) ); } #endif @@ -3513,6 +3558,7 @@ Rotate_log_event::Rotate_log_event(const char* buf, uint event_len, ident_offset = post_header_len; set_if_smaller(ident_len,FN_REFLEN-1); new_log_ident= my_strndup(buf + ident_offset, (uint) ident_len, MYF(MY_WME)); + DBUG_PRINT("debug", ("new_log_ident: '%s'", new_log_ident)); DBUG_VOID_RETURN; } @@ -3532,8 +3578,20 @@ bool Rotate_log_event::write(IO_CACHE* file) } #endif +/** + Helper function to detect if the event is inside a group. + */ +#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) +static bool is_in_group(THD *const thd, RELAY_LOG_INFO *const rli) +{ + return (thd->options & OPTION_BEGIN) != 0 || + (rli->last_event_start_time > 0); +} +#endif + + /* - Rotate_log_event::exec_event() + Rotate_log_event::do_apply_event() Got a rotate log event from the master @@ -3550,34 +3608,49 @@ bool Rotate_log_event::write(IO_CACHE* file) */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Rotate_log_event::exec_event(struct st_relay_log_info* rli) +int Rotate_log_event::do_update_pos(RELAY_LOG_INFO *rli) { - DBUG_ENTER("Rotate_log_event::exec_event"); + DBUG_ENTER("Rotate_log_event::do_update_pos"); +#ifndef DBUG_OFF + char buf[32]; +#endif + + DBUG_PRINT("info", ("server_id=%lu; ::server_id=%lu", + (ulong) this->server_id, (ulong) ::server_id)); + DBUG_PRINT("info", ("new_log_ident: %s", this->new_log_ident)); + DBUG_PRINT("info", ("pos: %s", llstr(this->pos, buf))); pthread_mutex_lock(&rli->data_lock); rli->event_relay_log_pos= my_b_tell(rli->cur_log); /* - If we are in a transaction: the only normal case is when the I/O thread was - copying a big transaction, then it was stopped and restarted: we have this - in the relay log: + If we are in a transaction or in a group: the only normal case is + when the I/O thread was copying a big transaction, then it was + stopped and restarted: we have this in the relay log: + BEGIN ... ROTATE (a fake one) ... COMMIT or ROLLBACK - In that case, we don't want to touch the coordinates which correspond to - the beginning of the transaction. - Starting from 5.0.0, there also are some rotates from the slave itself, in - the relay log. + + In that case, we don't want to touch the coordinates which + correspond to the beginning of the transaction. Starting from + 5.0.0, there also are some rotates from the slave itself, in the + relay log, which shall not change the group positions. */ - if (!(thd->options & OPTION_BEGIN)) + if ((server_id != ::server_id || rli->replicate_same_server_id) && + !is_in_group(thd, rli)) { + DBUG_PRINT("info", ("old group_master_log_name: '%s' " + "old group_master_log_pos: %lu", + rli->group_master_log_name, + (ulong) rli->group_master_log_pos)); memcpy(rli->group_master_log_name, new_log_ident, ident_len+1); rli->notify_group_master_log_name_update(); rli->group_master_log_pos= pos; rli->group_relay_log_pos= rli->event_relay_log_pos; - DBUG_PRINT("info", ("group_master_log_name: '%s' " - "group_master_log_pos: %lu", + DBUG_PRINT("info", ("new group_master_log_name: '%s' " + "new group_master_log_pos: %lu", rli->group_master_log_name, (ulong) rli->group_master_log_pos)); /* @@ -3596,8 +3669,28 @@ int Rotate_log_event::exec_event(struct st_relay_log_info* rli) pthread_mutex_unlock(&rli->data_lock); pthread_cond_broadcast(&rli->data_cond); flush_relay_log_info(rli); + DBUG_RETURN(0); } + + +Log_event::enum_skip_reason +Rotate_log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + enum_skip_reason reason= Log_event::do_shall_skip(rli); + + switch (reason) { + case Log_event::EVENT_SKIP_NOT: + case Log_event::EVENT_SKIP_COUNT: + return Log_event::EVENT_SKIP_NOT; + + case Log_event::EVENT_SKIP_IGNORE: + return Log_event::EVENT_SKIP_IGNORE; + } + DBUG_ASSERT(0); + return Log_event::EVENT_SKIP_NOT; // To keep compiler happy +} + #endif @@ -3704,11 +3797,11 @@ void Intvar_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) /* - Intvar_log_event::exec_event() + Intvar_log_event::do_apply_event() */ #if defined(HAVE_REPLICATION)&& !defined(MYSQL_CLIENT) -int Intvar_log_event::exec_event(struct st_relay_log_info* rli) +int Intvar_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { switch (type) { case LAST_INSERT_ID_EVENT: @@ -3719,9 +3812,33 @@ int Intvar_log_event::exec_event(struct st_relay_log_info* rli) thd->force_one_auto_inc_interval(val); break; } + return 0; +} + +int Intvar_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ rli->inc_event_relay_log_pos(); return 0; } + + +Log_event::enum_skip_reason +Intvar_log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + /* + It is a common error to set the slave skip counter to 1 instead of + 2 when recovering from an insert which used a auto increment, + rand, or user var. Therefore, if the slave skip counter is 1, we + just say that this event should be skipped by ignoring it, meaning + that we do not change the value of the slave skip counter since it + will be decreased by the following insert event. + */ + if (rli->slave_skip_counter == 1) + return Log_event::EVENT_SKIP_IGNORE; + else + return Log_event::do_shall_skip(rli); +} + #endif @@ -3784,13 +3901,37 @@ void Rand_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Rand_log_event::exec_event(struct st_relay_log_info* rli) +int Rand_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { thd->rand.seed1= (ulong) seed1; thd->rand.seed2= (ulong) seed2; + return 0; +} + +int Rand_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ rli->inc_event_relay_log_pos(); return 0; } + + +Log_event::enum_skip_reason +Rand_log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + /* + It is a common error to set the slave skip counter to 1 instead of + 2 when recovering from an insert which used a auto increment, + rand, or user var. Therefore, if the slave skip counter is 1, we + just say that this event should be skipped by ignoring it, meaning + that we do not change the value of the slave skip counter since it + will be decreased by the following insert event. + */ + if (rli->slave_skip_counter == 1) + return Log_event::EVENT_SKIP_IGNORE; + else + return Log_event::do_shall_skip(rli); +} + #endif /* !MYSQL_CLIENT */ @@ -3857,12 +3998,12 @@ void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Xid_log_event::exec_event(struct st_relay_log_info* rli) +int Xid_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { /* For a slave Xid_log_event is COMMIT */ general_log_print(thd, COM_QUERY, "COMMIT /* implicit, from Xid_log_event */"); - return end_trans(thd, COMMIT) || Log_event::exec_event(rli); + return end_trans(thd, COMMIT); } #endif /* !MYSQL_CLIENT */ @@ -4140,11 +4281,11 @@ void User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) /* - User_var_log_event::exec_event() + User_var_log_event::do_apply_event() */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int User_var_log_event::exec_event(struct st_relay_log_info* rli) +int User_var_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { Item *it= 0; CHARSET_INFO *charset; @@ -4206,9 +4347,31 @@ int User_var_log_event::exec_event(struct st_relay_log_info* rli) e.update_hash(val, val_len, type, charset, DERIVATION_IMPLICIT, 0); free_root(thd->mem_root,0); + return 0; +} + +int User_var_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ rli->inc_event_relay_log_pos(); return 0; } + +Log_event::enum_skip_reason +User_var_log_event::do_shall_skip(RELAY_LOG_INFO *rli) +{ + /* + It is a common error to set the slave skip counter to 1 instead + of 2 when recovering from an insert which used a auto increment, + rand, or user var. Therefore, if the slave skip counter is 1, we + just say that this event should be skipped by ignoring it, meaning + that we do not change the value of the slave skip counter since it + will be decreased by the following insert event. + */ + if (rli->slave_skip_counter == 1) + return Log_event::EVENT_SKIP_IGNORE; + else + return Log_event::do_shall_skip(rli); +} #endif /* !MYSQL_CLIENT */ @@ -4248,7 +4411,7 @@ void Slave_log_event::pack_info(Protocol *protocol) #ifndef MYSQL_CLIENT Slave_log_event::Slave_log_event(THD* thd_arg, - struct st_relay_log_info* rli) + RELAY_LOG_INFO* rli) :Log_event(thd_arg, 0, 0) , mem_pool(0), master_host(0) { DBUG_ENTER("Slave_log_event"); @@ -4358,11 +4521,11 @@ Slave_log_event::Slave_log_event(const char* buf, uint event_len) #ifndef MYSQL_CLIENT -int Slave_log_event::exec_event(struct st_relay_log_info* rli) +int Slave_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { if (mysql_bin_log.is_open()) mysql_bin_log.write(this); - return Log_event::exec_event(rli); + return 0; } #endif /* !MYSQL_CLIENT */ @@ -4391,21 +4554,21 @@ void Stop_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) /* - Stop_log_event::exec_event() - - The master stopped. - We used to clean up all temporary tables but this is useless as, as the - master has shut down properly, it has written all DROP TEMPORARY TABLE - (prepared statements' deletion is TODO only when we binlog prep stmts). - We used to clean up slave_load_tmpdir, but this is useless as it has been - cleared at the end of LOAD DATA INFILE. - So we have nothing to do here. - The place were we must do this cleaning is in Start_log_event_v3::exec_event(), - not here. Because if we come here, the master was sane. + Stop_log_event::do_apply_event() + + The master stopped. We used to clean up all temporary tables but + this is useless as, as the master has shut down properly, it has + written all DROP TEMPORARY TABLE (prepared statements' deletion is + TODO only when we binlog prep stmts). We used to clean up + slave_load_tmpdir, but this is useless as it has been cleared at the + end of LOAD DATA INFILE. So we have nothing to do here. The place + were we must do this cleaning is in + Start_log_event_v3::do_apply_event(), not here. Because if we come + here, the master was sane. */ #ifndef MYSQL_CLIENT -int Stop_log_event::exec_event(struct st_relay_log_info* rli) +int Stop_log_event::do_update_pos(RELAY_LOG_INFO *rli) { /* We do not want to update master_log pos because we get a rotate event @@ -4423,6 +4586,7 @@ int Stop_log_event::exec_event(struct st_relay_log_info* rli) } return 0; } + #endif /* !MYSQL_CLIENT */ #endif /* HAVE_REPLICATION */ @@ -4613,11 +4777,11 @@ void Create_file_log_event::pack_info(Protocol *protocol) /* - Create_file_log_event::exec_event() + Create_file_log_event::do_apply_event() */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Create_file_log_event::exec_event(struct st_relay_log_info* rli) +int Create_file_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { char proc_info[17+FN_REFLEN+10], *fname_buf; char *ext; @@ -4679,7 +4843,7 @@ err: if (fd >= 0) my_close(fd, MYF(0)); thd->proc_info= 0; - return error ? 1 : Log_event::exec_event(rli); + return error == 0; } #endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */ @@ -4787,15 +4951,15 @@ int Append_block_log_event::get_create_or_append() const } /* - Append_block_log_event::exec_event() + Append_block_log_event::do_apply_event() */ -int Append_block_log_event::exec_event(struct st_relay_log_info* rli) +int Append_block_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { char proc_info[17+FN_REFLEN+10], *fname= proc_info+17; int fd; int error = 1; - DBUG_ENTER("Append_block_log_event::exec_event"); + DBUG_ENTER("Append_block_log_event::do_apply_event"); fname= strmov(proc_info, "Making temp file "); slave_load_file_stem(fname, file_id, server_id, ".data"); @@ -4834,7 +4998,7 @@ err: if (fd >= 0) my_close(fd, MYF(0)); thd->proc_info= 0; - DBUG_RETURN(error ? error : Log_event::exec_event(rli)); + DBUG_RETURN(error); } #endif @@ -4918,18 +5082,18 @@ void Delete_file_log_event::pack_info(Protocol *protocol) #endif /* - Delete_file_log_event::exec_event() + Delete_file_log_event::do_apply_event() */ #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) -int Delete_file_log_event::exec_event(struct st_relay_log_info* rli) +int Delete_file_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { char fname[FN_REFLEN+10]; char *ext= slave_load_file_stem(fname, file_id, server_id, ".data"); (void) my_delete(fname, MYF(MY_WME)); strmov(ext, ".info"); (void) my_delete(fname, MYF(MY_WME)); - return Log_event::exec_event(rli); + return 0; } #endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */ @@ -5015,10 +5179,10 @@ void Execute_load_log_event::pack_info(Protocol *protocol) /* - Execute_load_log_event::exec_event() + Execute_load_log_event::do_apply_event() */ -int Execute_load_log_event::exec_event(struct st_relay_log_info* rli) +int Execute_load_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { char fname[FN_REFLEN+10]; char *ext; @@ -5049,14 +5213,15 @@ int Execute_load_log_event::exec_event(struct st_relay_log_info* rli) lev->thd = thd; /* - lev->exec_event should use rli only for errors - i.e. should not advance rli's position. - lev->exec_event is the place where the table is loaded (it calls - mysql_load()). + lev->do_apply_event should use rli only for errors i.e. should + not advance rli's position. + + lev->do_apply_event is the place where the table is loaded (it + calls mysql_load()). */ - rli->future_group_master_log_pos= log_pos; - if (lev->exec_event(0,rli,1)) + const_cast<RELAY_LOG_INFO*>(rli)->future_group_master_log_pos= log_pos; + if (lev->do_apply_event(0,rli,1)) { /* We want to indicate the name of the file that could not be loaded @@ -5099,7 +5264,7 @@ err: my_close(fd, MYF(0)); end_io_cache(&file); } - return error ? error : Log_event::exec_event(rli); + return error; } #endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */ @@ -5267,7 +5432,7 @@ void Execute_load_query_log_event::pack_info(Protocol *protocol) int -Execute_load_query_log_event::exec_event(struct st_relay_log_info* rli) +Execute_load_query_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { char *p; char *buf; @@ -5304,7 +5469,7 @@ Execute_load_query_log_event::exec_event(struct st_relay_log_info* rli) p= strmake(p, STRING_WITH_LEN(" INTO")); p= strmake(p, query+fn_pos_end, q_len-fn_pos_end); - error= Query_log_event::exec_event(rli, buf, p-buf); + error= Query_log_event::do_apply_event(rli, buf, p-buf); /* Forging file name for deletion in same buffer */ *fname_end= 0; @@ -5360,7 +5525,7 @@ bool sql_ex_info::write_data(IO_CACHE* file) sql_ex_info::init() */ -char* sql_ex_info::init(char* buf,char* buf_end,bool use_new_format) +char *sql_ex_info::init(char *buf, char *buf_end, bool use_new_format) { cached_new_format = use_new_format; if (use_new_format) @@ -5373,11 +5538,12 @@ char* sql_ex_info::init(char* buf,char* buf_end,bool use_new_format) the case when we have old format because we will be reusing net buffer to read the actual file before we write out the Create_file event. */ - if (read_str(&buf, buf_end, &field_term, &field_term_len) || - read_str(&buf, buf_end, &enclosed, &enclosed_len) || - read_str(&buf, buf_end, &line_term, &line_term_len) || - read_str(&buf, buf_end, &line_start, &line_start_len) || - read_str(&buf, buf_end, &escaped, &escaped_len)) + const char *ptr= buf; + if (read_str(&ptr, buf_end, (const char **) &field_term, &field_term_len) || + read_str(&ptr, buf_end, (const char **) &enclosed, &enclosed_len) || + read_str(&ptr, buf_end, (const char **) &line_term, &line_term_len) || + read_str(&ptr, buf_end, (const char **) &line_start, &line_start_len) || + read_str(&ptr, buf_end, (const char **) &escaped, &escaped_len)) return 0; opt_flags = *buf++; } @@ -5445,7 +5611,10 @@ Rows_log_event::Rows_log_event(THD *thd_arg, TABLE *tbl_arg, ulong tid, memcpy(m_cols.bitmap, cols->bitmap, no_bytes_in_map(cols)); } else - m_cols.bitmap= 0; // to not free it + { + // Needed because bitmap_init() does not set it to null on failure + m_cols.bitmap= 0; + } } #endif @@ -5482,14 +5651,57 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len, m_flags= uint2korr(post_start); - byte const *const var_start= (const byte *)buf + common_header_len + - post_header_len; + byte const *const var_start= + (const byte *)buf + common_header_len + post_header_len; byte const *const ptr_width= var_start; uchar *ptr_after_width= (uchar*) ptr_width; + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); m_width = net_field_length(&ptr_after_width); + DBUG_PRINT("debug", ("m_width=%lu", m_width)); + /* if bitmap_init fails, catched in is_valid() */ + if (likely(!bitmap_init(&m_cols, + m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL, + (m_width + 7) & ~7UL, + false))) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + memcpy(m_cols.bitmap, ptr_after_width, (m_width + 7) / 8); + ptr_after_width+= (m_width + 7) / 8; + DBUG_DUMP("m_cols", (char*) m_cols.bitmap, no_bytes_in_map(&m_cols)); + } + else + { + // Needed because bitmap_init() does not set it to null on failure + m_cols.bitmap= NULL; + DBUG_VOID_RETURN; + } + + m_cols_ai.bitmap= m_cols.bitmap; /* See explanation in is_valid() */ - const uint byte_count= (m_width + 7) / 8; - const byte* const ptr_rows_data= var_start + byte_count + 1; + if (event_type == UPDATE_ROWS_EVENT) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + + /* if bitmap_init fails, catched in is_valid() */ + if (likely(!bitmap_init(&m_cols_ai, + m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai : NULL, + (m_width + 7) & ~7UL, + false))) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + memcpy(m_cols_ai.bitmap, ptr_after_width, (m_width + 7) / 8); + ptr_after_width+= (m_width + 7) / 8; + DBUG_DUMP("m_cols_ai", (char*) m_cols_ai.bitmap, no_bytes_in_map(&m_cols_ai)); + } + else + { + // Needed because bitmap_init() does not set it to null on failure + m_cols_ai.bitmap= 0; + DBUG_VOID_RETURN; + } + } + + const byte* const ptr_rows_data= (const byte*) ptr_after_width; my_size_t const data_size= event_len - (ptr_rows_data - (const byte *) buf); DBUG_PRINT("info",("m_table_id: %lu m_flags: %d m_width: %lu data_size: %lu", @@ -5498,12 +5710,6 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len, m_rows_buf= (byte*)my_malloc(data_size, MYF(MY_WME)); if (likely((bool)m_rows_buf)) { - /* if bitmap_init fails, catched in is_valid() */ - if (likely(!bitmap_init(&m_cols, - m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL, - (m_width + 7) & ~7UL, - false))) - memcpy(m_cols.bitmap, ptr_after_width, byte_count); m_rows_end= m_rows_buf + data_size; m_rows_cur= m_rows_end; memcpy(m_rows_buf, ptr_rows_data, data_size); @@ -5522,6 +5728,29 @@ Rows_log_event::~Rows_log_event() my_free((gptr)m_rows_buf, MYF(MY_ALLOW_ZERO_PTR)); } +int Rows_log_event::get_data_size() +{ + int const type_code= get_type_code(); + + char buf[sizeof(m_width)+1]; + char *end= net_store_length(buf, (m_width + 7) / 8); + + DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master", + return 6 + no_bytes_in_map(&m_cols) + (end - buf) + + (type_code == UPDATE_ROWS_EVENT ? no_bytes_in_map(&m_cols_ai) : 0) + + (m_rows_cur - m_rows_buf);); + int data_size= ROWS_HEADER_LEN; + data_size+= no_bytes_in_map(&m_cols); + data_size+= end - buf; + + if (type_code == UPDATE_ROWS_EVENT) + data_size+= no_bytes_in_map(&m_cols_ai); + + data_size+= (m_rows_cur - m_rows_buf); + return data_size; +} + + #ifndef MYSQL_CLIENT int Rows_log_event::do_add_row_data(byte *const row_data, my_size_t const length) @@ -5595,10 +5824,10 @@ int Rows_log_event::do_add_row_data(byte *const row_data, row_end Pointer to variable that will hold the value of the one-after-end position for the row master_reclength - Pointer to variable that will be set to the length of the - record on the master side - rw_set Pointer to bitmap that holds either the read_set or the - write_set of the table + Pointer to variable that will be set to the length of the + record on the master side + rw_set Pointer to bitmap that holds either the read_set or the + write_set of the table DESCRIPTION @@ -5624,70 +5853,80 @@ int Rows_log_event::do_add_row_data(byte *const row_data, the master does not have a default value (and isn't nullable) */ static int -unpack_row(RELAY_LOG_INFO *rli, +unpack_row(RELAY_LOG_INFO const *rli, TABLE *table, uint const colcnt, - char const *row, MY_BITMAP const *cols, - char const **row_end, ulong *master_reclength, + char const *const row_data, MY_BITMAP const *cols, + char const **const row_end, ulong *const master_reclength, MY_BITMAP* const rw_set, Log_event_type const event_type) { - byte *const record= table->record[0]; DBUG_ENTER("unpack_row"); - DBUG_ASSERT(record && row); - DBUG_PRINT("enter", ("row: 0x%lx table->record[0]: 0x%lx", (long) row, (long) record)); - my_size_t master_null_bytes= table->s->null_bytes; - - if (colcnt != table->s->fields) - { - Field **fptr= &table->field[colcnt-1]; - do - master_null_bytes= (*fptr)->last_null_byte(); - while (master_null_bytes == Field::LAST_NULL_BYTE_UNDEF && - fptr-- > table->field); - - /* - If master_null_bytes is LAST_NULL_BYTE_UNDEF (0) at this time, - there were no nullable fields nor BIT fields at all in the - columns that are common to the master and the slave. In that - case, there is only one null byte holding the X bit. + DBUG_ASSERT(row_data); + my_size_t const master_null_byte_count= (bitmap_bits_set(cols) + 7) / 8; + int error= 0; - OBSERVE! There might still be nullable columns following the - common columns, so table->s->null_bytes might be greater than 1. - */ - if (master_null_bytes == Field::LAST_NULL_BYTE_UNDEF) - master_null_bytes= 1; - } + char const *null_ptr= row_data; + char const *pack_ptr= row_data + master_null_byte_count; - DBUG_ASSERT(master_null_bytes <= table->s->null_bytes); - memcpy(record, row, master_null_bytes); // [1] - int error= 0; + bitmap_clear_all(rw_set); - bitmap_set_all(rw_set); + empty_record(table); Field **const begin_ptr = table->field; Field **field_ptr; - char const *ptr= row + master_null_bytes; Field **const end_ptr= begin_ptr + colcnt; + + DBUG_ASSERT(null_ptr < row_data + master_null_byte_count); + + // Mask to mask out the correct bit among the null bits + unsigned int null_mask= 1U; + // The "current" null bits + unsigned int null_bits= *null_ptr++; for (field_ptr= begin_ptr ; field_ptr < end_ptr ; ++field_ptr) { Field *const f= *field_ptr; + /* + No need to bother about columns that does not exist: they have + gotten default values when being emptied above. + */ if (bitmap_is_set(cols, field_ptr - begin_ptr)) { - DBUG_ASSERT((const char *)table->record[0] <= f->ptr); - DBUG_ASSERT(f->ptr < (char*)(table->record[0] + table->s->reclength + - (f->pack_length_in_rec() == 0))); + if ((null_mask & 0xFF) == 0) + { + DBUG_ASSERT(null_ptr < row_data + master_null_byte_count); + null_mask= 1U; + null_bits= *null_ptr++; + } + + DBUG_ASSERT(null_mask & 0xFF); // One of the 8 LSB should be set - DBUG_PRINT("info", ("unpacking column '%s' to 0x%lx", f->field_name, - (long) f->ptr)); - ptr= f->unpack(f->ptr, ptr); /* Field...::unpack() cannot return 0 */ - DBUG_ASSERT(ptr != NULL); + DBUG_ASSERT(pack_ptr != NULL); + + if ((null_bits & null_mask) && f->maybe_null()) + f->set_null(); + else + { + f->set_notnull(); + + /* + We only unpack the field if it was non-null + */ + pack_ptr= f->unpack(f->ptr, pack_ptr); + } + + bitmap_set_bit(rw_set, f->field_index); + null_mask <<= 1; } - else - bitmap_clear_bit(rw_set, field_ptr - begin_ptr); } - *row_end = ptr; + /* + We should now have read all the null bytes, otherwise something is + really wrong. + */ + DBUG_ASSERT(null_ptr == row_data + master_null_byte_count); + + *row_end = pack_ptr; if (master_reclength) { if (*field_ptr) @@ -5712,9 +5951,8 @@ unpack_row(RELAY_LOG_INFO *rli, uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG; Field *const f= *field_ptr; - DBUG_PRINT("info", ("processing column '%s' @ 0x%lx", f->field_name, - (long) f->ptr)); - if (event_type == WRITE_ROWS_EVENT && (f->flags & mask) == mask) + if (event_type == WRITE_ROWS_EVENT && + ((*field_ptr)->flags & mask) == mask) { slave_print_msg(ERROR_LEVEL, rli, ER_NO_DEFAULT_FOR_FIELD, "Field `%s` of table `%s`.`%s` " @@ -5730,17 +5968,17 @@ unpack_row(RELAY_LOG_INFO *rli, DBUG_RETURN(error); } -int Rows_log_event::exec_event(st_relay_log_info *rli) +int Rows_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { - DBUG_ENTER("Rows_log_event::exec_event(st_relay_log_info*)"); + DBUG_ENTER("Rows_log_event::do_apply_event(st_relay_log_info*)"); int error= 0; char const *row_start= (char const *)m_rows_buf; /* - If m_table_id == ~0UL, then we have a dummy event that does - not contain any data. In that case, we just remove all tables in - the tables_to_lock list, close the thread tables, step the relay - log position, and return with success. + If m_table_id == ~0UL, then we have a dummy event that does not + contain any data. In that case, we just remove all tables in the + tables_to_lock list, close the thread tables, and return with + success. */ if (m_table_id == ~0UL) { @@ -5750,16 +5988,16 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) */ DBUG_ASSERT(get_flags(STMT_END_F)); - rli->clear_tables_to_lock(); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); close_thread_tables(thd); thd->clear_error(); - rli->inc_event_relay_log_pos(); DBUG_RETURN(0); } /* 'thd' has been set by exec_relay_log_event(), just before calling - exec_event(). We still check here to prevent future coding errors. + do_apply_event(). We still check here to prevent future coding + errors. */ DBUG_ASSERT(rli->sql_thd == thd); @@ -5775,8 +6013,9 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) /* lock_tables() reads the contents of thd->lex, so they must be - initialized. Contrary to in Table_map_log_event::exec_event() we don't - call mysql_init_query() as that may reset the binlog format. + initialized. Contrary to in + Table_map_log_event::do_apply_event() we don't call + mysql_init_query() as that may reset the binlog format. */ lex_start(thd, NULL, 0); @@ -5805,7 +6044,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) "Error in %s event: when locking tables", get_type_str()); } - rli->clear_tables_to_lock(); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); DBUG_RETURN(error); } @@ -5826,7 +6065,8 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) TABLE_LIST *tables= rli->tables_to_lock; close_tables_for_reopen(thd, &tables); - if ((error= open_tables(thd, &tables, &rli->tables_to_lock_count, 0))) + uint tables_count= rli->tables_to_lock_count; + if ((error= open_tables(thd, &tables, &tables_count, 0))) { if (thd->query_error || thd->is_fatal_error) { @@ -5841,7 +6081,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) "unexpected success or fatal error")); thd->query_error= 1; } - rli->clear_tables_to_lock(); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); DBUG_RETURN(error); } } @@ -5863,7 +6103,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) mysql_unlock_tables(thd, thd->lock); thd->lock= 0; thd->query_error= 1; - rli->clear_tables_to_lock(); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); DBUG_RETURN(ERR_BAD_TABLE_DEF); } } @@ -5885,24 +6125,24 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) */ for (TABLE_LIST *ptr= rli->tables_to_lock ; ptr ; ptr= ptr->next_global) { - rli->m_table_map.set_table(ptr->table_id, ptr->table); + const_cast<RELAY_LOG_INFO*>(rli)->m_table_map.set_table(ptr->table_id, ptr->table); } #ifdef HAVE_QUERY_CACHE query_cache.invalidate_locked_for_write(rli->tables_to_lock); #endif - rli->clear_tables_to_lock(); + const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock(); } DBUG_ASSERT(rli->tables_to_lock == NULL && rli->tables_to_lock_count == 0); - TABLE* table= rli->m_table_map.get_table(m_table_id); + TABLE* table= const_cast<RELAY_LOG_INFO*>(rli)->m_table_map.get_table(m_table_id); if (table) { /* table == NULL means that this table should not be replicated - (this was set up by Table_map_log_event::exec_event() which - tested replicate-* rules). + (this was set up by Table_map_log_event::do_apply_event() + which tested replicate-* rules). */ /* @@ -5959,7 +6199,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) break; default: - slave_print_msg(ERROR_LEVEL, rli, error, + slave_print_msg(ERROR_LEVEL, rli, thd->net.last_errno, "Error in %s event: row application failed", get_type_str()); thd->query_error= 1; @@ -5969,7 +6209,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) row_start= row_end; } DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event", - rli->abort_slave=1;); + const_cast<RELAY_LOG_INFO*>(rli)->abort_slave= 1;); error= do_after_row_operations(table, error); if (!cache_stmt) { @@ -5980,11 +6220,12 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) if (error) { /* error has occured during the transaction */ - slave_print_msg(ERROR_LEVEL, rli, error, + slave_print_msg(ERROR_LEVEL, rli, thd->net.last_errno, "Error in %s event: error during transaction execution " "on table %s.%s", get_type_str(), table->s->db.str, table->s->table_name.str); + /* If one day we honour --skip-slave-errors in row-based replication, and the error should be skipped, then we would clear mappings, rollback, @@ -5997,7 +6238,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) rollback at the caller along with sbr. */ thd->reset_current_stmt_binlog_row_based(); - rli->cleanup_context(thd, 0); /* rollback at caller in step with sbr */ + const_cast<RELAY_LOG_INFO*>(rli)->cleanup_context(thd, error); thd->query_error= 1; DBUG_RETURN(error); } @@ -6041,8 +6282,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) */ thd->reset_current_stmt_binlog_row_based(); - rli->cleanup_context(thd, 0); - rli->transaction_end(thd); + const_cast<RELAY_LOG_INFO*>(rli)->cleanup_context(thd, 0); if (error == 0) { @@ -6055,7 +6295,6 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) do not become visible. We still prefer to wipe them out. */ thd->clear_error(); - error= Log_event::exec_event(rli); } else slave_print_msg(ERROR_LEVEL, rli, error, @@ -6082,17 +6321,17 @@ int Rows_log_event::exec_event(st_relay_log_info *rli) wait (reached end of last relay log and nothing gets appended there), we timeout after one minute, and notify DBA about the problem. When WL#2975 is implemented, just remove the member - st_relay_log_info::unsafe_to_stop_at and all its occurences. + st_relay_log_info::last_event_start_time and all its occurences. */ - rli->unsafe_to_stop_at= time(0); + const_cast<RELAY_LOG_INFO*>(rli)->last_event_start_time= time(0); } DBUG_ASSERT(error == 0); thd->clear_error(); - rli->inc_event_relay_log_pos(); - + DBUG_RETURN(0); } + #endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */ #ifndef MYSQL_CLIENT @@ -6119,15 +6358,35 @@ bool Rows_log_event::write_data_body(IO_CACHE*file) */ char sbuf[sizeof(m_width)]; my_ptrdiff_t const data_size= m_rows_cur - m_rows_buf; + bool res= false; char *const sbuf_end= net_store_length((char*) sbuf, (uint) m_width); DBUG_ASSERT(static_cast<my_size_t>(sbuf_end - sbuf) <= sizeof(sbuf)); - return (my_b_safe_write(file, reinterpret_cast<byte*>(sbuf), - sbuf_end - sbuf) || - my_b_safe_write(file, reinterpret_cast<byte*>(m_cols.bitmap), - no_bytes_in_map(&m_cols)) || - my_b_safe_write(file, m_rows_buf, (uint) data_size)); + DBUG_DUMP("m_width", sbuf, sbuf_end - sbuf); + res= res || my_b_safe_write(file, + reinterpret_cast<byte*>(sbuf), + sbuf_end - sbuf); + + DBUG_DUMP("m_cols", (char*) m_cols.bitmap, no_bytes_in_map(&m_cols)); + res= res || my_b_safe_write(file, + reinterpret_cast<byte*>(m_cols.bitmap), + no_bytes_in_map(&m_cols)); + /* + TODO[refactor write]: Remove the "down cast" here (and elsewhere). + */ + if (get_type_code() == UPDATE_ROWS_EVENT) + { + DBUG_DUMP("m_cols_ai", (char*) m_cols_ai.bitmap, no_bytes_in_map(&m_cols_ai)); + res= res || my_b_safe_write(file, + reinterpret_cast<byte*>(m_cols_ai.bitmap), + no_bytes_in_map(&m_cols_ai)); + } + DBUG_DUMP("rows", m_rows_buf, data_size); + res= res || my_b_safe_write(file, m_rows_buf, (uint) data_size); + + return res; + } #endif @@ -6160,10 +6419,8 @@ void Rows_log_event::print_helper(FILE *file, if (get_flags(STMT_END_F)) { - my_b_copy_to_file(head, file); - my_b_copy_to_file(body, file); - reinit_io_cache(head, WRITE_CACHE, 0, FALSE, TRUE); - reinit_io_cache(body, WRITE_CACHE, 0, FALSE, TRUE); + copy_event_cache_to_file_and_reinit(head, file); + copy_event_cache_to_file_and_reinit(body, file); } } #endif @@ -6272,15 +6529,15 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len, const char *const vpart= buf + common_header_len + post_header_len; /* Extract the length of the various parts from the buffer */ - byte const* const ptr_dblen= (byte const*)vpart + 0; + byte const *const ptr_dblen= (byte const*)vpart + 0; m_dblen= *(uchar*) ptr_dblen; /* Length of database name + counter + terminating null */ - byte const* const ptr_tbllen= ptr_dblen + m_dblen + 2; + byte const *const ptr_tbllen= ptr_dblen + m_dblen + 2; m_tbllen= *(uchar*) ptr_tbllen; /* Length of table name + counter + terminating null */ - byte const* const ptr_colcnt= ptr_tbllen + m_tbllen + 2; + byte const *const ptr_colcnt= ptr_tbllen + m_tbllen + 2; uchar *ptr_after_colcnt= (uchar*) ptr_colcnt; m_colcnt= net_field_length(&ptr_after_colcnt); @@ -6325,9 +6582,9 @@ Table_map_log_event::~Table_map_log_event() */ #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) -int Table_map_log_event::exec_event(st_relay_log_info *rli) +int Table_map_log_event::do_apply_event(RELAY_LOG_INFO const *rli) { - DBUG_ENTER("Table_map_log_event::exec_event(st_relay_log_info*)"); + DBUG_ENTER("Table_map_log_event::do_apply_event(st_relay_log_info*)"); DBUG_ASSERT(rli->sql_thd == thd); @@ -6450,29 +6707,24 @@ int Table_map_log_event::exec_event(st_relay_log_info *rli) locked by linking the table into the list of tables to lock. */ table_list->next_global= table_list->next_local= rli->tables_to_lock; - rli->tables_to_lock= table_list; - rli->tables_to_lock_count++; + const_cast<RELAY_LOG_INFO*>(rli)->tables_to_lock= table_list; + const_cast<RELAY_LOG_INFO*>(rli)->tables_to_lock_count++; /* 'memory' is freed in clear_tables_to_lock */ } - /* - We explicitly do not call Log_event::exec_event() here since we do not - want the relay log position to be flushed to disk. The flushing will be - done by the last Rows_log_event that either ends a statement (outside a - transaction) or a transaction. - - A table map event can *never* end a transaction or a statement, so we - just step the relay log position. - */ - - if (likely(!error)) - rli->inc_event_relay_log_pos(); DBUG_RETURN(error); err: my_free((gptr) memory, MYF(MY_WME)); DBUG_RETURN(error); } + +int Table_map_log_event::do_update_pos(RELAY_LOG_INFO *rli) +{ + rli->inc_event_relay_log_pos(); + return 0; +} + #endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */ #ifndef MYSQL_CLIENT @@ -6637,7 +6889,7 @@ int Write_rows_log_event::do_after_row_operations(TABLE *table, int error) return error; } -int Write_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli, +int Write_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO const *rli, TABLE *table, char const *const row_start, char const **const row_end) @@ -6778,6 +7030,42 @@ copy_extra_record_fields(TABLE *table, return 0; // All OK } +#define DBUG_PRINT_BITSET(N,FRM,BS) \ + do { \ + char buf[256]; \ + for (uint i = 0 ; i < (BS)->n_bits ; ++i) \ + buf[i] = bitmap_is_set((BS), i) ? '1' : '0'; \ + buf[(BS)->n_bits] = '\0'; \ + DBUG_PRINT((N), ((FRM), buf)); \ + } while (0) + + +/** + Check if an error is a duplicate key error. + + This function is used to check if an error code is one of the + duplicate key error, i.e., and error code for which it is sensible + to do a <code>get_dup_key()</code> to retrieve the duplicate key. + + @param errcode The error code to check. + + @return <code>true</code> if the error code is such that + <code>get_dup_key()</code> will return true, <code>false</code> + otherwise. + */ +bool +is_duplicate_key_error(int errcode) +{ + switch (errcode) + { + case HA_ERR_FOUND_DUPP_KEY: + case HA_ERR_FOUND_DUPP_UNIQUE: + return true; + } + return false; +} + + /* Replace the provided record in the database. @@ -6810,6 +7098,12 @@ replace_record(THD *thd, TABLE *table, int keynum; auto_afree_ptr<char> key(NULL); +#ifndef DBUG_OFF + DBUG_DUMP("record[0]", table->record[0], table->s->reclength); + DBUG_PRINT_BITSET("debug", "write_set = %s", table->write_set); + DBUG_PRINT_BITSET("debug", "read_set = %s", table->read_set); +#endif + while ((error= table->file->ha_write_row(table->record[0]))) { if (error == HA_ERR_LOCK_DEADLOCK || error == HA_ERR_LOCK_WAIT_TIMEOUT) @@ -6820,7 +7114,7 @@ replace_record(THD *thd, TABLE *table, if ((keynum= table->file->get_dup_key(error)) < 0) { /* We failed to retrieve the duplicate key */ - DBUG_RETURN(HA_ERR_FOUND_DUPP_KEY); + DBUG_RETURN(error); } /* @@ -6837,7 +7131,10 @@ replace_record(THD *thd, TABLE *table, { error= table->file->rnd_pos(table->record[1], table->file->dup_ref); if (error) + { + table->file->print_error(error, MYF(0)); DBUG_RETURN(error); + } } else { @@ -6855,10 +7152,14 @@ replace_record(THD *thd, TABLE *table, key_copy((byte*)key.get(), table->record[0], table->key_info + keynum, 0); error= table->file->index_read_idx(table->record[1], keynum, - (const byte*)key.get(), HA_WHOLE_KEY, + (const byte*)key.get(), + HA_WHOLE_KEY, HA_READ_KEY_EXACT); if (error) + { + table->file->print_error(error, MYF(0)); DBUG_RETURN(error); + } } /* @@ -6891,15 +7192,21 @@ replace_record(THD *thd, TABLE *table, { error=table->file->ha_update_row(table->record[1], table->record[0]); + if (error) + table->file->print_error(error, MYF(0)); DBUG_RETURN(error); } else { if ((error= table->file->ha_delete_row(table->record[1]))) + { + table->file->print_error(error, MYF(0)); DBUG_RETURN(error); + } /* Will retry ha_write_row() with the offending row removed. */ } } + DBUG_RETURN(error); } @@ -6930,20 +7237,75 @@ void Write_rows_log_event::print(FILE *file, PRINT_EVENT_INFO* print_event_info) */ static bool record_compare(TABLE *table) { + /* + Need to set the X bit and the filler bits in both records since + there are engines that do not set it correctly. + + In addition, since MyISAM checks that one hasn't tampered with the + record, it is necessary to restore the old bytes into the record + after doing the comparison. + + TODO[record format ndb]: Remove it once NDB returns correct + records. Check that the other engines also return correct records. + */ + + bool result= FALSE; + byte saved_x[2], saved_filler[2]; + + if (table->s->null_bytes > 0) + { + for (int i = 0 ; i < 2 ; ++i) + { + saved_x[i]= table->record[i][0]; + saved_filler[i]= table->record[i][table->s->null_bytes - 1]; + table->record[i][0]|= 1U; + table->record[i][table->s->null_bytes - 1]|= + 256U - (1U << table->s->last_null_bit_pos); + } + } + if (table->s->blob_fields + table->s->varchar_fields == 0) - return cmp_record(table,record[1]); + { + result= cmp_record(table,record[1]); + goto record_compare_exit; + } + /* Compare null bits */ if (memcmp(table->null_flags, table->null_flags+table->s->rec_buff_length, table->s->null_bytes)) - return TRUE; // Diff in NULL value + { + result= TRUE; // Diff in NULL value + goto record_compare_exit; + } + /* Compare updated fields */ for (Field **ptr=table->field ; *ptr ; ptr++) { if ((*ptr)->cmp_binary_offset(table->s->rec_buff_length)) - return TRUE; + { + result= TRUE; + goto record_compare_exit; + } } - return FALSE; + +record_compare_exit: + /* + Restore the saved bytes. + + TODO[record format ndb]: Remove this code once NDB returns the + correct record format. + */ + if (table->s->null_bytes > 0) + { + for (int i = 0 ; i < 2 ; ++i) + { + table->record[i][0]= saved_x[i]; + table->record[i][table->s->null_bytes - 1]= saved_filler[i]; + } + } + + return result; } @@ -6979,6 +7341,8 @@ static int find_and_fetch_row(TABLE *table, byte *key) DBUG_ASSERT(table->in_use != NULL); + DBUG_DUMP("record[0]", table->record[0], table->s->reclength); + if ((table->file->ha_table_flags() & HA_PRIMARY_KEY_REQUIRED_FOR_POSITION) && table->s->primary_key < MAX_KEY) { @@ -7078,15 +7442,22 @@ static int find_and_fetch_row(TABLE *table, byte *key) while (record_compare(table)) { int error; + /* We need to set the null bytes to ensure that the filler bit are all set when returning. There are storage engines that just set the necessary bits on the bytes and don't set the filler bits correctly. + + TODO[record format ndb]: Remove this code once NDB returns the + correct record format. */ - my_ptrdiff_t const pos= - table->s->null_bytes > 0 ? table->s->null_bytes - 1 : 0; - table->record[1][pos]= 0xFF; + if (table->s->null_bytes > 0) + { + table->record[1][table->s->null_bytes - 1]|= + 256U - (1U << table->s->last_null_bit_pos); + } + if ((error= table->file->index_next(table->record[1]))) { table->file->print_error(error, MYF(0)); @@ -7112,17 +7483,11 @@ static int find_and_fetch_row(TABLE *table, byte *key) /* Continue until we find the right record or have made a full loop */ do { - /* - We need to set the null bytes to ensure that the filler bit - are all set when returning. There are storage engines that - just set the necessary bits on the bytes and don't set the - filler bits correctly. - */ - my_ptrdiff_t const pos= - table->s->null_bytes > 0 ? table->s->null_bytes - 1 : 0; - table->record[1][pos]= 0xFF; error= table->file->rnd_next(table->record[1]); + DBUG_DUMP("record[0]", table->record[0], table->s->reclength); + DBUG_DUMP("record[1]", table->record[1], table->s->reclength); + switch (error) { case 0: @@ -7136,6 +7501,7 @@ static int find_and_fetch_row(TABLE *table, byte *key) default: table->file->print_error(error, MYF(0)); + DBUG_PRINT("info", ("Record not found")); table->file->ha_rnd_end(); DBUG_RETURN(error); } @@ -7145,6 +7511,7 @@ static int find_and_fetch_row(TABLE *table, byte *key) /* Have to restart the scan to be able to fetch the next row. */ + DBUG_PRINT("info", ("Record %sfound", restart_count == 2 ? "not " : "")); table->file->ha_rnd_end(); DBUG_ASSERT(error == HA_ERR_END_OF_FILE || error == 0); @@ -7237,7 +7604,7 @@ int Delete_rows_log_event::do_after_row_operations(TABLE *table, int error) return error; } -int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli, +int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO const *rli, TABLE *table, char const *const row_start, char const **const row_end) @@ -7303,16 +7670,55 @@ void Delete_rows_log_event::print(FILE *file, */ #if !defined(MYSQL_CLIENT) Update_rows_log_event::Update_rows_log_event(THD *thd_arg, TABLE *tbl_arg, - ulong tid, MY_BITMAP const *cols, + ulong tid, + MY_BITMAP const *cols_bi, + MY_BITMAP const *cols_ai, + bool is_transactional) +: Rows_log_event(thd_arg, tbl_arg, tid, cols_bi, is_transactional) +#ifdef HAVE_REPLICATION + , m_memory(NULL), m_key(NULL) + +#endif +{ + init(cols_ai); +} + +Update_rows_log_event::Update_rows_log_event(THD *thd_arg, TABLE *tbl_arg, + ulong tid, + MY_BITMAP const *cols, bool is_transactional) : Rows_log_event(thd_arg, tbl_arg, tid, cols, is_transactional) #ifdef HAVE_REPLICATION , m_memory(NULL), m_key(NULL) #endif { + init(cols); +} + +void Update_rows_log_event::init(MY_BITMAP const *cols) +{ + /* if bitmap_init fails, catched in is_valid() */ + if (likely(!bitmap_init(&m_cols_ai, + m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai : NULL, + (m_width + 7) & ~7UL, + false))) + { + /* Cols can be zero if this is a dummy binrows event */ + if (likely(cols != NULL)) + memcpy(m_cols_ai.bitmap, cols->bitmap, no_bytes_in_map(cols)); + } } #endif /* !defined(MYSQL_CLIENT) */ + +Update_rows_log_event::~Update_rows_log_event() +{ + if (m_cols_ai.bitmap == m_bitbuf_ai) // no my_malloc happened + m_cols_ai.bitmap= 0; // so no my_free in bitmap_free + bitmap_free(&m_cols_ai); // To pair with bitmap_init(). +} + + /* Constructor used by slave to read the event from the binary log. */ @@ -7372,7 +7778,7 @@ int Update_rows_log_event::do_after_row_operations(TABLE *table, int error) return error; } -int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli, +int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO const *rli, TABLE *table, char const *const row_start, char const **const row_end) @@ -7397,7 +7803,7 @@ int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli, store_record(table, record[1]); char const *next_start = *row_end; /* m_after_image is the after image for the update */ - error= unpack_row(rli, table, m_width, next_start, &m_cols, row_end, + error= unpack_row(rli, table, m_width, next_start, &m_cols_ai, row_end, &m_master_reclength, table->write_set, UPDATE_ROWS_EVENT); bmove_align(m_after_image, table->record[0], table->s->reclength); restore_record(table, record[1]); @@ -7469,3 +7875,113 @@ void Update_rows_log_event::print(FILE *file, } #endif + +Incident_log_event::Incident_log_event(const char *buf, uint event_len, + const Format_description_log_event *descr_event) + : Log_event(buf, descr_event) +{ + DBUG_ENTER("Incident_log_event::Incident_log_event"); + uint8 const common_header_len= + descr_event->common_header_len; + uint8 const post_header_len= + descr_event->post_header_len[INCIDENT_EVENT-1]; + + DBUG_PRINT("info",("event_len: %u; common_header_len: %d; post_header_len: %d", + event_len, common_header_len, post_header_len)); + + m_incident= static_cast<Incident>(uint2korr(buf + common_header_len)); + char const *ptr= buf + common_header_len + post_header_len; + char const *const str_end= buf + event_len; + uint8 len= 0; // Assignment to keep compiler happy + const char *str= NULL; // Assignment to keep compiler happy + read_str(&ptr, str_end, &str, &len); + m_message.str= const_cast<char*>(str); + m_message.length= len; + DBUG_PRINT("info", ("m_incident: %d", m_incident)); + DBUG_VOID_RETURN; +} + + +Incident_log_event::~Incident_log_event() +{ +} + + +const char * +Incident_log_event::description() const +{ + static const char *const description[]= { + "NOTHING", // Not used + "LOST_EVENTS" + }; + + DBUG_PRINT("info", ("m_incident: %d", m_incident)); + + DBUG_ASSERT(0 <= m_incident); + DBUG_ASSERT((my_size_t) m_incident <= sizeof(description)/sizeof(*description)); + + return description[m_incident]; +} + + +#ifndef MYSQL_CLIENT +void Incident_log_event::pack_info(Protocol *protocol) +{ + char buf[256]; + my_size_t bytes; + if (m_message.length > 0) + bytes= my_snprintf(buf, sizeof(buf), "#%d (%s)", + m_incident, description()); + else + bytes= my_snprintf(buf, sizeof(buf), "#%d (%s): %s", + m_incident, description(), m_message.str); + protocol->store(buf, bytes, &my_charset_bin); +} +#endif + + +#ifdef MYSQL_CLIENT +void +Incident_log_event::print(FILE *file, + PRINT_EVENT_INFO *print_event_info) +{ + if (print_event_info->short_form) + return; + + Write_on_release_cache cache(&print_event_info->head_cache, file); + print_header(&cache, print_event_info, FALSE); + my_b_printf(&cache, "\n# Incident: %s", description()); +} +#endif + +#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) +int +Incident_log_event::do_apply_event(RELAY_LOG_INFO const *rli) +{ + DBUG_ENTER("Incident_log_event::do_apply_event"); + slave_print_msg(ERROR_LEVEL, rli, ER_SLAVE_INCIDENT, + ER(ER_SLAVE_INCIDENT), + description(), + m_message.length > 0 ? m_message.str : "<none>"); + DBUG_RETURN(1); +} +#endif + +bool +Incident_log_event::write_data_header(IO_CACHE *file) +{ + DBUG_ENTER("Incident_log_event::write_data_header"); + DBUG_PRINT("enter", ("m_incident: %d", m_incident)); + byte buf[sizeof(int16)]; + int2store(buf, (int16) m_incident); + DBUG_RETURN(my_b_safe_write(file, buf, sizeof(buf))); +} + +bool +Incident_log_event::write_data_body(IO_CACHE *file) +{ + DBUG_ENTER("Incident_log_event::write_data_body"); + DBUG_RETURN(write_str(file, m_message.str, m_message.length)); +} + + diff --git a/sql/log_event.h b/sql/log_event.h index 7cbe8925d9a..51543291621 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -22,6 +22,7 @@ #endif #include <my_bitmap.h> +#include "rpl_constants.h" #define LOG_READ_EOF -1 #define LOG_READ_BOGUS -2 @@ -198,7 +199,7 @@ struct sql_ex_info #define TABLE_MAP_HEADER_LEN 8 #define EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN (4 + 4 + 4 + 1) #define EXECUTE_LOAD_QUERY_HEADER_LEN (QUERY_HEADER_LEN + EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN) - +#define INCIDENT_HEADER_LEN 2 /* Max number of possible extra bytes in a replication event compared to a packet (i.e. a query) sent from client to master; @@ -468,10 +469,28 @@ enum Log_event_type XID_EVENT= 16, BEGIN_LOAD_QUERY_EVENT= 17, EXECUTE_LOAD_QUERY_EVENT= 18, + TABLE_MAP_EVENT = 19, - WRITE_ROWS_EVENT = 20, - UPDATE_ROWS_EVENT = 21, - DELETE_ROWS_EVENT = 22, + + /* + These event numbers were used for 5.1.0 to 5.1.15 and are + therefore obsolete. + */ + PRE_GA_WRITE_ROWS_EVENT = 20, + PRE_GA_UPDATE_ROWS_EVENT = 21, + PRE_GA_DELETE_ROWS_EVENT = 22, + + /* + These event numbers are used from 5.1.16 and forward + */ + WRITE_ROWS_EVENT = 23, + UPDATE_ROWS_EVENT = 24, + DELETE_ROWS_EVENT = 25, + + /* + Something out of the ordinary happened on the master + */ + INCIDENT_EVENT= 26, /* Add new events here - right above this comment! @@ -503,6 +522,7 @@ class THD; class Format_description_log_event; struct st_relay_log_info; +typedef st_relay_log_info RELAY_LOG_INFO; #ifdef MYSQL_CLIENT /* @@ -591,6 +611,33 @@ typedef struct st_print_event_info class Log_event { public: + /** + Enumeration of what kinds of skipping (and non-skipping) that can + occur when the slave executes an event. + + @see shall_skip + @see do_shall_skip + */ + enum enum_skip_reason { + /** + Don't skip event. + */ + EVENT_SKIP_NOT, + + /** + Skip event by ignoring it. + + This means that the slave skip counter will not be changed. + */ + EVENT_SKIP_IGNORE, + + /** + Skip event and decrease skip counter. + */ + EVENT_SKIP_COUNT + }; + + /* The following type definition is to be used whenever data is placed and manipulated in a common buffer. Use this typedef for buffers @@ -672,16 +719,14 @@ public: static void init_show_field_list(List<Item>* field_list); #ifdef HAVE_REPLICATION int net_send(Protocol *protocol, const char* log_name, my_off_t pos); + /* pack_info() is used by SHOW BINLOG EVENTS; as print() it prepares and sends a string to display to the user, so it resembles print(). */ + virtual void pack_info(Protocol *protocol); - /* - The SQL slave thread calls exec_event() to execute the event; this is where - the slave's data is modified. - */ - virtual int exec_event(struct st_relay_log_info* rli); + #endif /* HAVE_REPLICATION */ virtual const char* get_db() { @@ -754,6 +799,127 @@ public: *description_event); /* returns the human readable name of the event's type */ const char* get_type_str(); + +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) +public: + + /** + Apply the event to the database. + + This function represents the public interface for applying an + event. + + @see do_apply_event + */ + int apply_event(RELAY_LOG_INFO const *rli) { + return do_apply_event(rli); + } + + + /** + Update the relay log position. + + This function represents the public interface for "stepping over" + the event and will update the relay log information. + + @see do_update_pos + */ + int update_pos(RELAY_LOG_INFO *rli) + { + return do_update_pos(rli); + } + + /** + Decide if the event shall be skipped, and the reason for skipping + it. + + @see do_shall_skip + */ + enum_skip_reason shall_skip(RELAY_LOG_INFO *rli) + { + return do_shall_skip(rli); + } + +protected: + /** + Primitive to apply an event to the database. + + This is where the change to the database is made. + + @note The primitive is protected instead of private, since there + is a hierarchy of actions to be performed in some cases. + + @see Format_description_log_event::do_apply_event() + + @param rli Pointer to relay log info structure + + @retval 0 Event applied successfully + @retval errno Error code if event application failed + */ + virtual int do_apply_event(RELAY_LOG_INFO const *rli) + { + return 0; /* Default implementation does nothing */ + } + + + /** + Advance relay log coordinates. + + This function is called to advance the relay log coordinates to + just after the event. It is essential that both the relay log + coordinate and the group log position is updated correctly, since + this function is used also for skipping events. + + Normally, each implementation of do_update_pos() shall: + + - Update the event position to refer to the position just after + the event. + + - Update the group log position to refer to the position just + after the event <em>if the event is last in a group</em> + + @param rli Pointer to relay log info structure + + @retval 0 Coordinates changed successfully + @retval errno Error code if advancing failed (usually just + 1). Observe that handler errors are returned by the + do_apply_event() function, and not by this one. + */ + virtual int do_update_pos(RELAY_LOG_INFO *rli); + + + /** + Decide if this event shall be skipped or not and the reason for + skipping it. + + The default implementation decide that the event shall be skipped + if either: + + - the server id of the event is the same as the server id of the + server and <code>rli->replicate_same_server_id</code> is true, + or + + - if <code>rli->slave_skip_counter</code> is greater than zero. + + @see do_apply_event + @see do_update_pos + + @retval Log_event::EVENT_SKIP_NOT + The event shall not be skipped and should be applied. + + @retval Log_event::EVENT_SKIP_IGNORE + The event shall be skipped by just ignoring it, i.e., the slave + skip counter shall not be changed. This happends if, for example, + the originating server id of the event is the same as the server + id of the slave. + + @retval Log_event::EVENT_SKIP_COUNT + The event shall be skipped because the slave skip counter was + non-zero. The caller shall decrease the counter by one. + */ + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); + +#endif }; /* @@ -794,10 +960,10 @@ public: uint16 error_code; ulong thread_id; /* - For events created by Query_log_event::exec_event (and - Load_log_event::exec_event()) we need the *original* thread id, to be able - to log the event with the original (=master's) thread id (fix for - BUG#1686). + For events created by Query_log_event::do_apply_event (and + Load_log_event::do_apply_event()) we need the *original* thread + id, to be able to log the event with the original (=master's) + thread id (fix for BUG#1686). */ ulong slave_proxy_id; @@ -860,9 +1026,6 @@ public: const char* get_db() { return db; } #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); - int exec_event(struct st_relay_log_info* rli, const char *query_arg, - uint32 q_len_arg); #endif /* HAVE_REPLICATION */ #else void print_query_header(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info); @@ -891,6 +1054,16 @@ public: */ virtual ulong get_post_header_size_for_derived() { return 0; } /* Writes derived event-specific part of post header. */ + +public: /* !!! Public in this patch to allow old usage */ +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); + + int do_apply_event(RELAY_LOG_INFO const *rli, + const char *query_arg, + uint32 q_len_arg); +#endif /* HAVE_REPLICATION */ }; @@ -939,9 +1112,8 @@ public: uint16 master_port; #ifndef MYSQL_CLIENT - Slave_log_event(THD* thd_arg, struct st_relay_log_info* rli); + Slave_log_event(THD* thd_arg, RELAY_LOG_INFO* rli); void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); #endif @@ -954,6 +1126,11 @@ public: #ifndef MYSQL_CLIENT bool write(IO_CACHE* file); #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const* rli); +#endif }; #endif /* HAVE_REPLICATION */ @@ -1023,12 +1200,6 @@ public: const char* get_db() { return db; } #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli) - { - return exec_event(thd->slave_net,rli,0); - } - int exec_event(NET* net, struct st_relay_log_info* rli, - bool use_rli_only_for_errors); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1060,6 +1231,17 @@ public: + LOAD_HEADER_LEN + sql_ex.data_size() + field_block_len + num_fields); } + +public: /* !!! Public in this patch to allow old usage */ +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const* rli) + { + return do_apply_event(thd->slave_net,rli,0); + } + + int do_apply_event(NET *net, RELAY_LOG_INFO const *rli, + bool use_rli_only_for_errors); +#endif }; extern char server_version[SERVER_VERSION_LENGTH]; @@ -1117,7 +1299,6 @@ public: Start_log_event_v3(); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else Start_log_event_v3() {} @@ -1137,6 +1318,22 @@ public: return START_V3_HEADER_LEN; //no variable-sized part } virtual bool is_artificial_event() { return artificial_event; } + +protected: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO*) + { + /* + Events from ourself should be skipped, but they should not + decrease the slave skip counter. + */ + if (this->server_id == ::server_id) + return Log_event::EVENT_SKIP_IGNORE; + else + return Log_event::EVENT_SKIP_NOT; + } +#endif }; @@ -1162,13 +1359,6 @@ public: uchar server_version_split[3]; Format_description_log_event(uint8 binlog_ver, const char* server_ver=0); - -#ifndef MYSQL_CLIENT -#ifdef HAVE_REPLICATION - int exec_event(struct st_relay_log_info* rli); -#endif /* HAVE_REPLICATION */ -#endif - Format_description_log_event(const char* buf, uint event_len, const Format_description_log_event* description_event); ~Format_description_log_event() { my_free((gptr)post_header_len, MYF(0)); } @@ -1191,7 +1381,15 @@ public: */ return FORMAT_DESCRIPTION_HEADER_LEN; } + void calc_server_version_split(); + +protected: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); +#endif }; @@ -1215,7 +1413,6 @@ public: {} #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1230,6 +1427,13 @@ public: bool write(IO_CACHE* file); #endif bool is_valid() const { return 1; } + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); +#endif }; @@ -1256,7 +1460,6 @@ class Rand_log_event: public Log_event {} #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1270,6 +1473,13 @@ class Rand_log_event: public Log_event bool write(IO_CACHE* file); #endif bool is_valid() const { return 1; } + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); +#endif }; /***************************************************************************** @@ -1293,7 +1503,6 @@ class Xid_log_event: public Log_event Xid_log_event(THD* thd_arg, my_xid x): Log_event(thd_arg,0,0), xid(x) {} #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1307,6 +1516,11 @@ class Xid_log_event: public Log_event bool write(IO_CACHE* file); #endif bool is_valid() const { return 1; } + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif }; /***************************************************************************** @@ -1336,7 +1550,6 @@ public: val_len(val_len_arg), type(type_arg), charset_number(charset_number_arg) { is_null= !val; } void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); #endif @@ -1348,6 +1561,13 @@ public: bool write(IO_CACHE* file); #endif bool is_valid() const { return 1; } + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); +#endif }; @@ -1362,7 +1582,6 @@ public: #ifndef MYSQL_CLIENT Stop_log_event() :Log_event() {} - int exec_event(struct st_relay_log_info* rli); #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); #endif @@ -1373,6 +1592,22 @@ public: ~Stop_log_event() {} Log_event_type get_type_code() { return STOP_EVENT;} bool is_valid() const { return 1; } + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli) + { + /* + Events from ourself should be skipped, but they should not + decrease the slave skip counter. + */ + if (this->server_id == ::server_id) + return Log_event::EVENT_SKIP_IGNORE; + else + return Log_event::EVENT_SKIP_NOT; + } +#endif }; /***************************************************************************** @@ -1399,7 +1634,6 @@ public: ulonglong pos_arg, uint flags); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1418,6 +1652,12 @@ public: #ifndef MYSQL_CLIENT bool write(IO_CACHE* file); #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_update_pos(RELAY_LOG_INFO *rli); + virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli); +#endif }; @@ -1452,7 +1692,6 @@ public: bool using_trans); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1486,6 +1725,11 @@ public: */ bool write_base(IO_CACHE* file); #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif }; @@ -1518,7 +1762,6 @@ public: Append_block_log_event(THD* thd, const char* db_arg, char* block_arg, uint block_len_arg, bool using_trans); #ifdef HAVE_REPLICATION - int exec_event(struct st_relay_log_info* rli); void pack_info(Protocol* protocol); virtual int get_create_or_append() const; #endif /* HAVE_REPLICATION */ @@ -1536,6 +1779,11 @@ public: bool write(IO_CACHE* file); const char* get_db() { return db; } #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif }; @@ -1555,7 +1803,6 @@ public: Delete_file_log_event(THD* thd, const char* db_arg, bool using_trans); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1572,6 +1819,11 @@ public: bool write(IO_CACHE* file); const char* get_db() { return db; } #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif }; @@ -1591,7 +1843,6 @@ public: Execute_load_log_event(THD* thd, const char* db_arg, bool using_trans); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1607,6 +1858,11 @@ public: bool write(IO_CACHE* file); const char* get_db() { return db; } #endif + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif }; @@ -1676,7 +1932,6 @@ public: bool using_trans, bool suppress_use); #ifdef HAVE_REPLICATION void pack_info(Protocol* protocol); - int exec_event(struct st_relay_log_info* rli); #endif /* HAVE_REPLICATION */ #else void print(FILE* file, PRINT_EVENT_INFO* print_event_info); @@ -1695,7 +1950,12 @@ public: #ifndef MYSQL_CLIENT bool write_post_header_for_derived(IO_CACHE* file); #endif - }; + +private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif +}; #ifdef MYSQL_CLIENT @@ -1793,7 +2053,6 @@ public: #endif #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) - virtual int exec_event(struct st_relay_log_info *rli); virtual void pack_info(Protocol *protocol); #endif @@ -1803,6 +2062,11 @@ public: private: +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + virtual int do_update_pos(RELAY_LOG_INFO *rli); +#endif + #ifndef MYSQL_CLIENT TABLE *m_table; #endif @@ -1886,7 +2150,6 @@ public: flag_set get_flags(flag_set flags) const { return m_flags & flags; } #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) - virtual int exec_event(struct st_relay_log_info *rli); virtual void pack_info(Protocol *protocol); #endif @@ -1903,14 +2166,7 @@ public: #endif /* Member functions to implement superclass interface */ - virtual int get_data_size() - { - DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master", - return 6 + 1 + no_bytes_in_map(&m_cols) + - (m_rows_cur - m_rows_buf);); - return ROWS_HEADER_LEN + 1 + no_bytes_in_map(&m_cols) + - (m_rows_cur - m_rows_buf); - } + virtual int get_data_size(); MY_BITMAP const *get_cols() const { return &m_cols; } my_size_t get_width() const { return m_width; } @@ -1921,9 +2177,14 @@ public: virtual bool write_data_body(IO_CACHE *file); virtual const char *get_db() { return m_table->s->db.str; } #endif + /* + Check that malloc() succeeded in allocating memory for the rows + buffer and the COLS vector. Checking that an Update_rows_log_event + is valid is done in the Update_rows_log_event::is_valid() + function. + */ virtual bool is_valid() const { - /* that's how we check malloc() succeeded */ return m_rows_buf && m_cols.bitmap; } @@ -1956,10 +2217,20 @@ protected: ulong m_table_id; /* Table ID */ MY_BITMAP m_cols; /* Bitmap denoting columns available */ ulong m_width; /* The width of the columns bitmap */ + /* + Bitmap for columns available in the after image, if present. These + fields are only available for Update_rows events. Observe that the + width of both the before image COLS vector and the after image + COLS vector is the same: the number of columns of the table on the + master. + */ + MY_BITMAP m_cols_ai; + ulong m_master_reclength; /* Length of record on master side */ - /* Bit buffer in the same memory as the class */ + /* Bit buffers in the same memory as the class */ uint32 m_bitbuf[128/(sizeof(uint32)*8)]; + uint32 m_bitbuf_ai[128/(sizeof(uint32)*8)]; byte *m_rows_buf; /* The rows in packed format */ byte *m_rows_cur; /* One-after the end of the data */ @@ -1970,6 +2241,8 @@ protected: private: #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); + /* Primitive to prepare for a sequence of row executions. @@ -2017,7 +2290,7 @@ private: RETURN VALUE Error code, if something went wrong, 0 otherwise. */ - virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*, + virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*, char const *row_start, char const **row_end) = 0; /* @@ -2088,7 +2361,7 @@ private: 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*, + virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*, char const *row_start, char const **row_end); virtual int do_exec_row(TABLE *table); #endif @@ -2117,10 +2390,20 @@ public: }; #ifndef MYSQL_CLIENT - Update_rows_log_event(THD*, TABLE*, ulong table_id, - MY_BITMAP const *cols, bool is_transactional); + Update_rows_log_event(THD*, TABLE*, ulong table_id, + MY_BITMAP const *cols_bi, + MY_BITMAP const *cols_ai, + bool is_transactional); + + Update_rows_log_event(THD*, TABLE*, ulong table_id, + MY_BITMAP const *cols, + bool is_transactional); + + void init(MY_BITMAP const *cols); #endif + virtual ~Update_rows_log_event(); + #ifdef HAVE_REPLICATION Update_rows_log_event(const char *buf, uint event_len, const Format_description_log_event *description_event); @@ -2139,6 +2422,11 @@ public: } #endif + virtual bool is_valid() const + { + return Rows_log_event::is_valid() && m_cols_ai.bitmap; + } + private: virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; } @@ -2153,7 +2441,7 @@ private: 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*, + virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*, char const *row_start, char const **row_end); virtual int do_exec_row(TABLE *table); #endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */ @@ -2224,10 +2512,108 @@ private: 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*, + virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*, char const *row_start, char const **row_end); virtual int do_exec_row(TABLE *table); #endif }; + +/** + Class representing an incident, an occurance out of the ordinary, + that happened on the master. + + The event is used to inform the slave that something out of the + ordinary happened on the master that might cause the database to be + in an inconsistent state. + + <table id="IncidentFormat"> + <caption>Incident event format</caption> + <tr> + <th>Symbol</th> + <th>Size<br/>(bytes)</th> + <th>Description</th> + </tr> + <tr> + <td>INCIDENT</td> + <td align="right">2</td> + <td>Incident number as an unsigned integer</td> + </tr> + <tr> + <td>MSGLEN</td> + <td align="right">1</td> + <td>Message length as an unsigned integer</td> + </tr> + <tr> + <td>MESSAGE</td> + <td align="right">MSGLEN</td> + <td>The message, if present. Not null terminated.</td> + </tr> + </table> + */ +class Incident_log_event : public Log_event { +public: +#ifndef MYSQL_CLIENT + Incident_log_event(THD *thd_arg, Incident incident) + : Log_event(thd_arg, 0, FALSE), m_incident(incident) + { + DBUG_ENTER("Incident_log_event::Incident_log_event"); + DBUG_PRINT("enter", ("m_incident: %d", m_incident)); + m_message.str= NULL; /* Just as a precaution */ + m_message.length= 0; + DBUG_VOID_RETURN; + } + + Incident_log_event(THD *thd_arg, Incident incident, LEX_STRING const msg) + : Log_event(thd_arg, 0, FALSE), m_incident(incident) + { + DBUG_ENTER("Incident_log_event::Incident_log_event"); + DBUG_PRINT("enter", ("m_incident: %d", m_incident)); + m_message= msg; + DBUG_VOID_RETURN; + } +#endif + +#ifndef MYSQL_CLIENT + void pack_info(Protocol*); +#endif + + Incident_log_event(const char *buf, uint event_len, + const Format_description_log_event *descr_event); + + virtual ~Incident_log_event(); + +#ifdef MYSQL_CLIENT + virtual void print(FILE *file, PRINT_EVENT_INFO *print_event_info); +#endif + +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + virtual int do_apply_event(RELAY_LOG_INFO const *rli); +#endif + + virtual bool write_data_header(IO_CACHE *file); + virtual bool write_data_body(IO_CACHE *file); + + virtual Log_event_type get_type_code() { return INCIDENT_EVENT; } + + virtual bool is_valid() const { return 1; } + virtual int get_data_size() { + return INCIDENT_HEADER_LEN + 1 + m_message.length; + } + +private: + const char *description() const; + + Incident m_incident; + LEX_STRING m_message; +}; + +static inline bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, + FILE *file) +{ + return + my_b_copy_to_file(cache, file) || + reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE); +} + #endif /* _log_event_h */ diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h index 0dcdea5c4f4..958cd7f7838 100644 --- a/sql/mysql_priv.h +++ b/sql/mysql_priv.h @@ -1445,6 +1445,9 @@ void close_system_tables(THD *thd, Open_tables_state *backup); TABLE *open_system_table_for_update(THD *thd, TABLE_LIST *one_table); bool close_cached_tables(THD *thd, bool wait_for_refresh, TABLE_LIST *tables, bool have_lock = FALSE); +bool close_cached_connection_tables(THD *thd, bool wait_for_refresh, + LEX_STRING *connect_string, + bool have_lock = FALSE); void copy_field_from_tmp_record(Field *field,int offset); bool fill_record(THD *thd, Field **field, List<Item> &values, bool ignore_errors); diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 696937e4e55..8ad9fc1b74f 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -6273,7 +6273,7 @@ The minimum value for this variable is 4096.", (gptr*) &global_system_variables.net_write_timeout, (gptr*) &max_system_variables.net_write_timeout, 0, GET_ULONG, REQUIRED_ARG, NET_WRITE_TIMEOUT, 1, LONG_TIMEOUT, 0, 1, 0}, - { "old_mode", OPT_OLD_MODE, "Use compatible behaviour.", + { "old", OPT_OLD_MODE, "Use compatible behavior.", (gptr*) &global_system_variables.old_mode, (gptr*) &max_system_variables.old_mode, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0}, @@ -6456,12 +6456,12 @@ The minimum value for this variable is 4096.", (gptr*) &max_system_variables.tmp_table_size, 0, GET_ULL, REQUIRED_ARG, 16*1024*1024L, 1024, MAX_MEM_TABLE_SIZE, 0, 1, 0}, {"transaction_alloc_block_size", OPT_TRANS_ALLOC_BLOCK_SIZE, - "Allocation block size for transactions to be stored in binary log", + "Allocation block size for various transaction-related structures", (gptr*) &global_system_variables.trans_alloc_block_size, (gptr*) &max_system_variables.trans_alloc_block_size, 0, GET_ULONG, REQUIRED_ARG, QUERY_ALLOC_BLOCK_SIZE, 1024, ~0L, 0, 1024, 0}, {"transaction_prealloc_size", OPT_TRANS_PREALLOC_SIZE, - "Persistent buffer for transactions to be stored in binary log", + "Persistent buffer for various transaction-related structures", (gptr*) &global_system_variables.trans_prealloc_size, (gptr*) &max_system_variables.trans_prealloc_size, 0, GET_ULONG, REQUIRED_ARG, TRANS_ALLOC_PREALLOC_SIZE, 1024, ~0L, 0, 1024, 0}, diff --git a/sql/opt_range.cc b/sql/opt_range.cc index 03d97a294ad..f5cf79bd609 100644 --- a/sql/opt_range.cc +++ b/sql/opt_range.cc @@ -84,7 +84,7 @@ static int sel_cmp(Field *f,char *a,char *b,uint8 a_flag,uint8 b_flag); static char is_null_string[2]= {1,0}; - +class RANGE_OPT_PARAM; /* A construction block of the SEL_ARG-graph. @@ -170,6 +170,89 @@ static char is_null_string[2]= {1,0}; - get_quick_select() - Walk the SEL_ARG, materialize the key intervals, and create QUICK_RANGE_SELECT object that will read records within these intervals. + + 4. SPACE COMPLEXITY NOTES + + SEL_ARG graph is a representation of an ordered disjoint sequence of + intervals over the ordered set of index tuple values. + + For multi-part keys, one can construct a WHERE expression such that its + list of intervals will be of combinatorial size. Here is an example: + + (keypart1 IN (1,2, ..., n1)) AND + (keypart2 IN (1,2, ..., n2)) AND + (keypart3 IN (1,2, ..., n3)) + + For this WHERE clause the list of intervals will have n1*n2*n3 intervals + of form + + (keypart1, keypart2, keypart3) = (k1, k2, k3), where 1 <= k{i} <= n{i} + + SEL_ARG graph structure aims to reduce the amount of required space by + "sharing" the elementary intervals when possible (the pic at the + beginning of this comment has examples of such sharing). The sharing may + prevent combinatorial blowup: + + There are WHERE clauses that have combinatorial-size interval lists but + will be represented by a compact SEL_ARG graph. + Example: + (keypartN IN (1,2, ..., n1)) AND + ... + (keypart2 IN (1,2, ..., n2)) AND + (keypart1 IN (1,2, ..., n3)) + + but not in all cases: + + - There are WHERE clauses that do have a compact SEL_ARG-graph + representation but get_mm_tree() and its callees will construct a + graph of combinatorial size. + Example: + (keypart1 IN (1,2, ..., n1)) AND + (keypart2 IN (1,2, ..., n2)) AND + ... + (keypartN IN (1,2, ..., n3)) + + - There are WHERE clauses for which the minimal possible SEL_ARG graph + representation will have combinatorial size. + Example: + By induction: Let's take any interval on some keypart in the middle: + + kp15=c0 + + Then let's AND it with this interval 'structure' from preceding and + following keyparts: + + (kp14=c1 AND kp16=c3) OR keypart14=c2) (*) + + We will obtain this SEL_ARG graph: + + kp14 $ kp15 $ kp16 + $ $ + +---------+ $ +---------+ $ +---------+ + | kp14=c1 |--$-->| kp15=c0 |--$-->| kp16=c3 | + +---------+ $ +---------+ $ +---------+ + | $ $ + +---------+ $ +---------+ $ + | kp14=c2 |--$-->| kp15=c0 | $ + +---------+ $ +---------+ $ + $ $ + + Note that we had to duplicate "kp15=c0" and there was no way to avoid + that. + The induction step: AND the obtained expression with another "wrapping" + expression like (*). + When the process ends because of the limit on max. number of keyparts + we'll have: + + WHERE clause length is O(3*#max_keyparts) + SEL_ARG graph size is O(2^(#max_keyparts/2)) + + (it is also possible to construct a case where instead of 2 in 2^n we + have a bigger constant, e.g. 4, and get a graph with 4^(31/2)= 2^31 + nodes) + + We avoid consuming too much memory by setting a limit on the number of + SEL_ARG object we can construct during one range analysis invocation. */ class SEL_ARG :public Sql_alloc @@ -200,6 +283,8 @@ public: enum leaf_color { BLACK,RED } color; enum Type { IMPOSSIBLE, MAYBE, MAYBE_KEY, KEY_RANGE } type; + enum { MAX_SEL_ARGS = 16000 }; + SEL_ARG() {} SEL_ARG(SEL_ARG &); SEL_ARG(Field *,const char *,const char *); @@ -271,7 +356,7 @@ public: return new SEL_ARG(field, part, min_value, arg->max_value, min_flag, arg->max_flag, maybe_flag | arg->maybe_flag); } - SEL_ARG *clone(SEL_ARG *new_parent,SEL_ARG **next); + SEL_ARG *clone(RANGE_OPT_PARAM *param, SEL_ARG *new_parent, SEL_ARG **next); bool copy_min(SEL_ARG* arg) { // Get overlapping range @@ -423,7 +508,6 @@ public: { return parent->left == this ? &parent->left : &parent->right; } - SEL_ARG *clone_tree(); /* @@ -466,6 +550,7 @@ public: } return !field->key_cmp(min_val, max_val); } + SEL_ARG *clone_tree(RANGE_OPT_PARAM *param); }; class SEL_IMERGE; @@ -545,6 +630,8 @@ public: using_real_indexes==TRUE */ uint real_keynr[MAX_KEY]; + /* Number of SEL_ARG objects allocated by SEL_ARG::clone_tree operations */ + uint alloced_sel_args; }; class PARAM : public RANGE_OPT_PARAM @@ -634,8 +721,9 @@ static void print_quick(QUICK_SELECT_I *quick, const key_map *needed_reg); static SEL_TREE *tree_and(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2); static SEL_TREE *tree_or(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2); static SEL_ARG *sel_add(SEL_ARG *key1,SEL_ARG *key2); -static SEL_ARG *key_or(SEL_ARG *key1,SEL_ARG *key2); -static SEL_ARG *key_and(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag); +static SEL_ARG *key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1, SEL_ARG *key2); +static SEL_ARG *key_and(RANGE_OPT_PARAM *param, SEL_ARG *key1, SEL_ARG *key2, + uint clone_flag); static bool get_range(SEL_ARG **e1,SEL_ARG **e2,SEL_ARG *root1); bool get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key, SEL_ARG *key_tree,char *min_key,uint min_key_flag, @@ -860,6 +948,7 @@ int imerge_list_or_tree(RANGE_OPT_PARAM *param, return im1->is_empty(); } + /*************************************************************************** ** Basic functions for SQL_SELECT and QUICK_RANGE_SELECT ***************************************************************************/ @@ -1510,12 +1599,18 @@ SEL_ARG::SEL_ARG(Field *field_,uint8 part_,char *min_value_,char *max_value_, left=right= &null_element; } -SEL_ARG *SEL_ARG::clone(SEL_ARG *new_parent,SEL_ARG **next_arg) +SEL_ARG *SEL_ARG::clone(RANGE_OPT_PARAM *param, SEL_ARG *new_parent, + SEL_ARG **next_arg) { SEL_ARG *tmp; + + /* Bail out if we have already generated too many SEL_ARGs */ + if (++param->alloced_sel_args > MAX_SEL_ARGS) + return 0; + if (type != KEY_RANGE) { - if (!(tmp= new SEL_ARG(type))) + if (!(tmp= new (param->mem_root) SEL_ARG(type))) return 0; // out of memory tmp->prev= *next_arg; // Link into next/prev chain (*next_arg)->next=tmp; @@ -1523,20 +1618,21 @@ SEL_ARG *SEL_ARG::clone(SEL_ARG *new_parent,SEL_ARG **next_arg) } else { - if (!(tmp= new SEL_ARG(field,part, min_value,max_value, - min_flag, max_flag, maybe_flag))) + if (!(tmp= new (param->mem_root) SEL_ARG(field,part, min_value,max_value, + min_flag, max_flag, maybe_flag))) return 0; // OOM tmp->parent=new_parent; tmp->next_key_part=next_key_part; if (left != &null_element) - tmp->left=left->clone(tmp,next_arg); + if (!(tmp->left=left->clone(param, tmp, next_arg))) + return 0; // OOM tmp->prev= *next_arg; // Link into next/prev chain (*next_arg)->next=tmp; (*next_arg)= tmp; if (right != &null_element) - if (!(tmp->right= right->clone(tmp,next_arg))) + if (!(tmp->right= right->clone(param, tmp, next_arg))) return 0; // OOM } increment_use_count(1); @@ -1614,11 +1710,12 @@ static int sel_cmp(Field *field, char *a,char *b,uint8 a_flag,uint8 b_flag) } -SEL_ARG *SEL_ARG::clone_tree() +SEL_ARG *SEL_ARG::clone_tree(RANGE_OPT_PARAM *param) { SEL_ARG tmp_link,*next_arg,*root; next_arg= &tmp_link; - root= clone((SEL_ARG *) 0, &next_arg); + if (!(root= clone(param, (SEL_ARG *) 0, &next_arg))) + return 0; next_arg->next=0; // Fix last link tmp_link.next->prev=0; // Fix first link if (root) // If not OOM @@ -2120,6 +2217,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use, param.real_keynr[param.keys++]=idx; } param.key_parts_end=key_parts; + param.alloced_sel_args= 0; /* Calculate cost of full index read for the shortest covering index */ if (!head->covering_keys.is_clear_all()) @@ -2512,6 +2610,7 @@ bool prune_partitions(THD *thd, TABLE *table, Item *pprune_cond) range_par->using_real_indexes= FALSE; range_par->remove_jump_scans= FALSE; range_par->real_keynr[0]= 0; + range_par->alloced_sel_args= 0; thd->no_errors=1; // Don't warn about NULL thd->mem_root=&alloc; @@ -5242,7 +5341,8 @@ static SEL_TREE *get_mm_tree(RANGE_OPT_PARAM *param,COND *cond) while ((item=li++)) { SEL_TREE *new_tree=get_mm_tree(param,item); - if (param->thd->is_fatal_error) + if (param->thd->is_fatal_error || + param->alloced_sel_args > SEL_ARG::MAX_SEL_ARGS) DBUG_RETURN(0); // out of memory tree=tree_and(param,tree,new_tree); if (tree && tree->type == SEL_TREE::IMPOSSIBLE) @@ -5818,9 +5918,9 @@ tree_and(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2) tree1->type=SEL_TREE::KEY_SMALLER; DBUG_RETURN(tree1); } - key_map result_keys; result_keys.clear_all(); + /* Join the trees key per key */ SEL_ARG **key1,**key2,**end; for (key1= tree1->keys,key2= tree2->keys,end=key1+param->keys ; @@ -5833,7 +5933,7 @@ tree_and(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2) flag|=CLONE_KEY1_MAYBE; if (*key2 && !(*key2)->simple_key()) flag|=CLONE_KEY2_MAYBE; - *key1=key_and(*key1,*key2,flag); + *key1=key_and(param, *key1, *key2, flag); if (*key1 && (*key1)->type == SEL_ARG::IMPOSSIBLE) { tree1->type= SEL_TREE::IMPOSSIBLE; @@ -5841,8 +5941,8 @@ tree_and(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2) } result_keys.set_bit(key1 - tree1->keys); #ifdef EXTRA_DEBUG - if (*key1) - (*key1)->test_use_count(*key1); + if (*key1 && param->alloced_sel_args < SEL_ARG::MAX_SEL_ARGS) + (*key1)->test_use_count(*key1); #endif } } @@ -5994,13 +6094,14 @@ tree_or(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2) for (key1= tree1->keys,key2= tree2->keys,end= key1+param->keys ; key1 != end ; key1++,key2++) { - *key1=key_or(*key1,*key2); + *key1=key_or(param, *key1, *key2); if (*key1) { result=tree1; // Added to tree1 result_keys.set_bit(key1 - tree1->keys); #ifdef EXTRA_DEBUG - (*key1)->test_use_count(*key1); + if (param->alloced_sel_args < SEL_ARG::MAX_SEL_ARGS) + (*key1)->test_use_count(*key1); #endif } } @@ -6058,14 +6159,15 @@ tree_or(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2) /* And key trees where key1->part < key2 -> part */ static SEL_ARG * -and_all_keys(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag) +and_all_keys(RANGE_OPT_PARAM *param, SEL_ARG *key1, SEL_ARG *key2, + uint clone_flag) { SEL_ARG *next; ulong use_count=key1->use_count; if (key1->elements != 1) { - key2->use_count+=key1->elements-1; + key2->use_count+=key1->elements-1; //psergey: why we don't count that key1 has n-k-p? key2->increment_use_count((int) key1->elements-1); } if (key1->type == SEL_ARG::MAYBE_KEY) @@ -6077,7 +6179,7 @@ and_all_keys(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag) { if (next->next_key_part) { - SEL_ARG *tmp=key_and(next->next_key_part,key2,clone_flag); + SEL_ARG *tmp= key_and(param, next->next_key_part, key2, clone_flag); if (tmp && tmp->type == SEL_ARG::IMPOSSIBLE) { key1=key1->tree_delete(next); @@ -6086,6 +6188,8 @@ and_all_keys(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag) next->next_key_part=tmp; if (use_count) next->increment_use_count(use_count); + if (param->alloced_sel_args > SEL_ARG::MAX_SEL_ARGS) + break; } else next->next_key_part=key2; @@ -6102,8 +6206,10 @@ and_all_keys(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag) SYNOPSIS key_and() - key1 First argument, root of its RB-tree - key2 Second argument, root of its RB-tree + param Range analysis context (needed to track if we have allocated + too many SEL_ARGs) + key1 First argument, root of its RB-tree + key2 Second argument, root of its RB-tree RETURN RB-tree root of the resulting SEL_ARG graph. @@ -6111,7 +6217,7 @@ and_all_keys(SEL_ARG *key1,SEL_ARG *key2,uint clone_flag) */ static SEL_ARG * -key_and(SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) +key_and(RANGE_OPT_PARAM *param, SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) { if (!key1) return key2; @@ -6127,9 +6233,9 @@ key_and(SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) // key1->part < key2->part key1->use_count--; if (key1->use_count > 0) - if (!(key1= key1->clone_tree())) + if (!(key1= key1->clone_tree(param))) return 0; // OOM - return and_all_keys(key1,key2,clone_flag); + return and_all_keys(param, key1, key2, clone_flag); } if (((clone_flag & CLONE_KEY2_MAYBE) && @@ -6147,14 +6253,14 @@ key_and(SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) if (key1->use_count > 1) { key1->use_count--; - if (!(key1=key1->clone_tree())) + if (!(key1=key1->clone_tree(param))) return 0; // OOM key1->use_count++; } if (key1->type == SEL_ARG::MAYBE_KEY) { // Both are maybe key - key1->next_key_part=key_and(key1->next_key_part,key2->next_key_part, - clone_flag); + key1->next_key_part=key_and(param, key1->next_key_part, + key2->next_key_part, clone_flag); if (key1->next_key_part && key1->next_key_part->type == SEL_ARG::IMPOSSIBLE) return key1; @@ -6165,7 +6271,7 @@ key_and(SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) if (key2->next_key_part) { key1->use_count--; // Incremented in and_all_keys - return and_all_keys(key1,key2,clone_flag); + return and_all_keys(param, key1, key2, clone_flag); } key2->use_count--; // Key2 doesn't have a tree } @@ -6201,7 +6307,8 @@ key_and(SEL_ARG *key1, SEL_ARG *key2, uint clone_flag) } else if (get_range(&e2,&e1,key2)) continue; - SEL_ARG *next=key_and(e1->next_key_part,e2->next_key_part,clone_flag); + SEL_ARG *next=key_and(param, e1->next_key_part, e2->next_key_part, + clone_flag); e1->increment_use_count(1); e2->increment_use_count(1); if (!next || next->type != SEL_ARG::IMPOSSIBLE) @@ -6249,7 +6356,7 @@ get_range(SEL_ARG **e1,SEL_ARG **e2,SEL_ARG *root1) static SEL_ARG * -key_or(SEL_ARG *key1,SEL_ARG *key2) +key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2) { if (!key1) { @@ -6297,7 +6404,7 @@ key_or(SEL_ARG *key1,SEL_ARG *key2) { swap_variables(SEL_ARG *,key1,key2); } - if (key1->use_count > 0 || !(key1=key1->clone_tree())) + if (key1->use_count > 0 || !(key1=key1->clone_tree(param))) return 0; // OOM } @@ -6441,7 +6548,7 @@ key_or(SEL_ARG *key1,SEL_ARG *key2) { // tmp.min. <= x <= tmp.max tmp->maybe_flag|= key.maybe_flag; key.increment_use_count(key1->use_count+1); - tmp->next_key_part=key_or(tmp->next_key_part,key.next_key_part); + tmp->next_key_part= key_or(param, tmp->next_key_part, key.next_key_part); if (!cmp) // Key2 is ready break; key.copy_max_to_min(tmp); @@ -6472,7 +6579,7 @@ key_or(SEL_ARG *key1,SEL_ARG *key2) tmp->increment_use_count(key1->use_count+1); /* Increment key count as it may be used for next loop */ key.increment_use_count(1); - new_arg->next_key_part=key_or(tmp->next_key_part,key.next_key_part); + new_arg->next_key_part= key_or(param, tmp->next_key_part, key.next_key_part); key1=key1->insert(new_arg); break; } diff --git a/sql/records.cc b/sql/records.cc index 0923ab1d75e..0fb9f4f9650 100644 --- a/sql/records.cc +++ b/sql/records.cc @@ -150,7 +150,8 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table, info->file= table->file; info->forms= &info->table; /* Only one table */ - if (table->s->tmp_table == TMP_TABLE && !table->sort.addon_field) + if (table->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE && + !table->sort.addon_field) VOID(table->file->extra(HA_EXTRA_MMAP)); if (table->sort.addon_field) diff --git a/sql/rpl_constants.h b/sql/rpl_constants.h new file mode 100644 index 00000000000..426e80a328d --- /dev/null +++ b/sql/rpl_constants.h @@ -0,0 +1,18 @@ +#ifndef RPL_CONSTANTS_H +#define RPL_CONSTANTS_H + +/** + Enumeration of the incidents that can occur for the server. + */ +enum Incident { + /** No incident */ + INCIDENT_NONE, + + /** There are possibly lost events in the replication stream */ + INCIDENT_LOST_EVENTS, + + /** Shall be last event of the enumeration */ + INCIDENT_COUNT +}; + +#endif /* RPL_CONSTANTS_H */ diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc index 95b5ecba895..b22b052a105 100644 --- a/sql/rpl_injector.cc +++ b/sql/rpl_injector.cc @@ -188,3 +188,21 @@ void injector::new_trans(THD *thd, injector::transaction *ptr) DBUG_VOID_RETURN; } + +int injector::record_incident(THD *thd, Incident incident) +{ + Incident_log_event ev(thd, incident); + if (int error= mysql_bin_log.write(&ev)) + return error; + mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE); + return 0; +} + +int injector::record_incident(THD *thd, Incident incident, LEX_STRING message) +{ + Incident_log_event ev(thd, incident, message); + if (int error= mysql_bin_log.write(&ev)) + return error; + mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE); + return 0; +} diff --git a/sql/rpl_injector.h b/sql/rpl_injector.h index 61c2e0ecebc..eabf374857a 100644 --- a/sql/rpl_injector.h +++ b/sql/rpl_injector.h @@ -18,9 +18,10 @@ /* Pull in 'byte', 'my_off_t', and 'uint32' */ #include <my_global.h> - #include <my_bitmap.h> +#include "rpl_constants.h" + /* Forward declarations */ class handler; class MYSQL_BIN_LOG; @@ -322,6 +323,9 @@ public: transaction new_trans(THD *); void new_trans(THD *, transaction *); + int record_incident(THD*, Incident incident); + int record_incident(THD*, Incident incident, LEX_STRING message); + private: explicit injector(); ~injector() { } /* Nothing needs to be done */ diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 8a051195dba..b0db355154e 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -29,14 +29,15 @@ int init_strvar_from_file(char *var, int max_size, IO_CACHE *f, st_relay_log_info::st_relay_log_info() - :no_storage(FALSE), info_fd(-1), cur_log_fd(-1), save_temporary_tables(0), + :no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id), + info_fd(-1), cur_log_fd(-1), save_temporary_tables(0), cur_log_old_open_count(0), group_master_log_pos(0), log_space_total(0), ignore_log_space_limit(0), last_master_timestamp(0), slave_skip_counter(0), abort_pos_wait(0), slave_run_id(0), sql_thd(0), last_slave_errno(0), inited(0), abort_slave(0), slave_running(0), until_condition(UNTIL_NONE), until_log_pos(0), retried_trans(0), tables_to_lock(0), tables_to_lock_count(0), - unsafe_to_stop_at(0) + last_event_start_time(0) { DBUG_ENTER("st_relay_log_info::st_relay_log_info"); @@ -968,7 +969,7 @@ err: strtol() conversions needed for log names comparison. We don't need to compare them each time this function is called, we only need to do this when current log name changes. If we have UNTIL_MASTER_POS condition we - need to do this only after Rotate_log_event::exec_event() (which is + need to do this only after Rotate_log_event::do_apply_event() (which is rare, so caching gives real benifit), and if we have UNTIL_RELAY_POS condition then we should invalidate cached comarison value after inc_group_relay_log_pos() which called for each group of events (so we @@ -1001,6 +1002,22 @@ bool st_relay_log_info::is_until_satisfied() log_pos= group_relay_log_pos; } +#ifndef DBUG_OFF + { + char buf[32]; + DBUG_PRINT("info", ("group_master_log_name='%s', group_master_log_pos=%s", + group_master_log_name, llstr(group_master_log_pos, buf))); + DBUG_PRINT("info", ("group_relay_log_name='%s', group_relay_log_pos=%s", + group_relay_log_name, llstr(group_relay_log_pos, buf))); + DBUG_PRINT("info", ("(%s) log_name='%s', log_pos=%s", + until_condition == UNTIL_MASTER_POS ? "master" : "relay", + log_name, llstr(log_pos, buf))); + DBUG_PRINT("info", ("(%s) until_log_name='%s', until_log_pos=%s", + until_condition == UNTIL_MASTER_POS ? "master" : "relay", + until_log_name, llstr(until_log_pos, buf))); + } +#endif + if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN) { /* @@ -1056,30 +1073,19 @@ void st_relay_log_info::cached_charset_invalidate() } -bool st_relay_log_info::cached_charset_compare(char *charset) +bool st_relay_log_info::cached_charset_compare(char *charset) const { DBUG_ENTER("st_relay_log_info::cached_charset_compare"); if (bcmp(cached_charset, charset, sizeof(cached_charset))) { - memcpy(cached_charset, charset, sizeof(cached_charset)); + memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset)); DBUG_RETURN(1); } DBUG_RETURN(0); } -void st_relay_log_info::transaction_end(THD* thd) -{ - DBUG_ENTER("st_relay_log_info::transaction_end"); - - /* - Nothing to do here right now. - */ - - DBUG_VOID_RETURN; -} - #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) void st_relay_log_info::cleanup_context(THD *thd, bool error) { @@ -1087,12 +1093,12 @@ void st_relay_log_info::cleanup_context(THD *thd, bool error) DBUG_ASSERT(sql_thd == thd); /* - 1) Instances of Table_map_log_event, if ::exec_event() was called on them, + 1) Instances of Table_map_log_event, if ::do_apply_event() was called on them, may have opened tables, which we cannot be sure have been closed (because maybe the Rows_log_event have not been found or will not be, because slave SQL thread is stopping, or relay log has a missing tail etc). So we close all thread's tables. And so the table mappings have to be cancelled. - 2) Rows_log_event::exec_event() may even have started statements or + 2) Rows_log_event::do_apply_event() may even have started statements or transactions on them, which we need to rollback in case of error. 3) If finding a Format_description_log_event after a BEGIN, we also need to rollback before continuing with the next events. @@ -1106,7 +1112,7 @@ void st_relay_log_info::cleanup_context(THD *thd, bool error) m_table_map.clear_tables(); close_thread_tables(thd); clear_tables_to_lock(); - unsafe_to_stop_at= 0; + last_event_start_time= 0; DBUG_VOID_RETURN; } diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index 45c9fb1cf96..3f06e108f6d 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -58,6 +58,15 @@ typedef struct st_relay_log_info */ bool no_storage; + /* + If true, events with the same server id should be replicated. This + field is set on creation of a relay log info structure by copying + the value of ::replicate_same_server_id and can be overridden if + necessary. For example of when this is done, check sql_binlog.cc, + where the BINLOG statement can be used to execute "raw" events. + */ + bool replicate_same_server_id; + /*** The following variables can only be read when protect by data lock ****/ /* @@ -292,14 +301,19 @@ typedef struct st_relay_log_info When the 6 bytes are equal to 0 is used to mean "cache is invalidated". */ void cached_charset_invalidate(); - bool cached_charset_compare(char *charset); - - void transaction_end(THD*); + bool cached_charset_compare(char *charset) const; void cleanup_context(THD *, bool); void clear_tables_to_lock(); - time_t unsafe_to_stop_at; + /* + Used by row-based replication to detect that it should not stop at + this event, but give it a chance to send more events. The time + where the last event inside a group started is stored here. If the + variable is zero, we are not in a group (but may be in a + transaction). + */ + time_t last_event_start_time; } RELAY_LOG_INFO; diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc index 65a44a4947b..1d7cc808f0c 100644 --- a/sql/rpl_utility.cc +++ b/sql/rpl_utility.cc @@ -108,7 +108,7 @@ field_length_from_packed(enum_field_types const field_type, */ int -table_def::compatible_with(RELAY_LOG_INFO *rli, TABLE *table) +table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table) const { /* @@ -116,6 +116,7 @@ table_def::compatible_with(RELAY_LOG_INFO *rli, TABLE *table) */ uint const cols_to_check= min(table->s->fields, size()); int error= 0; + RELAY_LOG_INFO const *rli= const_cast<RELAY_LOG_INFO*>(rli_arg); TABLE_SHARE const *const tsh= table->s; diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h index b1aa642619c..17879a9ecfc 100644 --- a/sql/rpl_utility.h +++ b/sql/rpl_utility.h @@ -117,7 +117,7 @@ public: @retval 1 if the table definition is not compatible with @c table @retval 0 if the table definition is compatible with @c table */ - int compatible_with(RELAY_LOG_INFO *rli, TABLE *table) const; + int compatible_with(RELAY_LOG_INFO const *rli, TABLE *table) const; private: my_size_t m_size; // Number of elements in the types array diff --git a/sql/set_var.cc b/sql/set_var.cc index fef6fccfd06..7f55556c134 100644 --- a/sql/set_var.cc +++ b/sql/set_var.cc @@ -356,6 +356,8 @@ sys_var_thd_ulong sys_net_retry_count("net_retry_count", &SV::net_retry_count, 0, fix_net_retry_count); sys_var_thd_bool sys_new_mode("new", &SV::new_mode); +sys_var_bool_ptr_readonly sys_old_mode("old", + &global_system_variables.old_mode); sys_var_thd_bool sys_old_alter_table("old_alter_table", &SV::old_alter_table); sys_var_thd_bool sys_old_passwords("old_passwords", &SV::old_passwords); @@ -942,6 +944,7 @@ SHOW_VAR init_vars[]= { {sys_net_retry_count.name, (char*) &sys_net_retry_count, SHOW_SYS}, {sys_net_write_timeout.name,(char*) &sys_net_write_timeout, SHOW_SYS}, {sys_new_mode.name, (char*) &sys_new_mode, SHOW_SYS}, + {sys_old_mode.name, (char*) &sys_old_mode, SHOW_SYS}, {sys_old_alter_table.name, (char*) &sys_old_alter_table, SHOW_SYS}, {sys_old_passwords.name, (char*) &sys_old_passwords, SHOW_SYS}, {"open_files_limit", (char*) &open_files_limit, SHOW_LONG}, diff --git a/sql/set_var.h b/sql/set_var.h index 8887d91ec69..eac03fce610 100644 --- a/sql/set_var.h +++ b/sql/set_var.h @@ -169,6 +169,16 @@ public: }; +class sys_var_bool_ptr_readonly :public sys_var_bool_ptr +{ +public: + sys_var_bool_ptr_readonly(const char *name_arg, my_bool *value_arg) + :sys_var_bool_ptr(name_arg, value_arg) + {} + bool is_readonly() const { return 1; } +}; + + class sys_var_str :public sys_var { public: diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt index 8159f014e93..a42f94760ab 100644 --- a/sql/share/errmsg.txt +++ b/sql/share/errmsg.txt @@ -6055,3 +6055,5 @@ ER_EVENT_CANNOT_CREATE_IN_THE_PAST eng "Event execution time is in the past and ON COMPLETION NOT PRESERVE is set. Event has not been created" ER_EVENT_CANNOT_ALTER_IN_THE_PAST eng "Event execution time is in the past and ON COMPLETION NOT PRESERVE is set. Event has not been altered" +ER_SLAVE_INCIDENT + eng "The incident %s occured on the master. Message: %-.64s" diff --git a/sql/slave.cc b/sql/slave.cc index 1622f0f93a2..bc6cef95fc6 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -522,11 +522,11 @@ static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli) really one minute of idleness, we don't timeout if the slave SQL thread is actively working. */ - if (!rli->unsafe_to_stop_at) + if (rli->last_event_start_time == 0) DBUG_RETURN(1); DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving " "it some grace period")); - if (difftime(time(0), rli->unsafe_to_stop_at) > 60) + if (difftime(time(0), rli->last_event_start_time) > 60) { slave_print_msg(ERROR_LEVEL, rli, 0, "SQL thread had to stop in an unsafe situation, in " @@ -560,7 +560,7 @@ static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli) void */ -void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli, +void slave_print_msg(enum loglevel level, RELAY_LOG_INFO const *rli, int err_code, const char* msg, ...) { void (*report_function)(const char *, ...); @@ -582,9 +582,9 @@ void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli, It's an error, it must be reported in Last_error and Last_errno in SHOW SLAVE STATUS. */ - pbuff= rli->last_slave_error; + pbuff= const_cast<RELAY_LOG_INFO*>(rli)->last_slave_error; pbuffsize= sizeof(rli->last_slave_error); - rli->last_slave_errno = err_code; + const_cast<RELAY_LOG_INFO*>(rli)->last_slave_errno = err_code; report_function= sql_print_error; break; case WARNING_LEVEL: @@ -816,7 +816,7 @@ do not trust column Seconds_Behind_Master of SHOW SLAVE STATUS"); { if ((master_row= mysql_fetch_row(master_res)) && (::server_id == strtoul(master_row[1], 0, 10)) && - !replicate_same_server_id) + !mi->rli.replicate_same_server_id) errmsg= "The slave I/O thread stops because master and slave have equal \ MySQL server ids; these ids must be different for replication to work (or \ the --replicate-same-server-id option must be used on slave but this does \ @@ -1393,7 +1393,7 @@ void set_slave_thread_options(THD* thd) DBUG_VOID_RETURN; } -void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli) +void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO const *rli) { DBUG_ENTER("set_slave_thread_default_charset"); @@ -1404,7 +1404,14 @@ void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli) thd->variables.collation_server= global_system_variables.collation_server; thd->update_charset(); - rli->cached_charset_invalidate(); + + /* + We use a const cast here since the conceptual (and externally + visible) behavior of the function is to set the default charset of + the thread. That the cache has to be invalidated is a secondary + effect. + */ + const_cast<RELAY_LOG_INFO*>(rli)->cached_charset_invalidate(); DBUG_VOID_RETURN; } @@ -1612,7 +1619,8 @@ static ulong read_event(MYSQL* mysql, MASTER_INFO *mi, bool* suppress_warnings) } -int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int expected_error) +int check_expected_error(THD* thd, RELAY_LOG_INFO const *rli, + int expected_error) { DBUG_ENTER("check_expected_error"); @@ -1717,78 +1725,43 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) } if (ev) { - int type_code = ev->get_type_code(); - int exec_res; + int const type_code= ev->get_type_code(); + int exec_res= 0; /* - Queries originating from this server must be skipped. - Low-level events (Format_desc, Rotate, Stop) from this server - must also be skipped. But for those we don't want to modify - group_master_log_pos, because these events did not exist on the master. - Format_desc is not completely skipped. - Skip queries specified by the user in slave_skip_counter. - We can't however skip events that has something to do with the - log files themselves. - Filtering on own server id is extremely important, to ignore execution of - events created by the creation/rotation of the relay log (remember that - now the relay log starts with its Format_desc, has a Rotate etc). */ - DBUG_PRINT("info",("type_code=%d, server_id=%d",type_code,ev->server_id)); + DBUG_PRINT("info",("type_code=%d (%s), server_id=%d", + type_code, ev->get_type_str(), ev->server_id)); + DBUG_PRINT("info", ("thd->options={ %s%s}", + FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT), + FLAGSTR(thd->options, OPTION_BEGIN))); - if ((ev->server_id == (uint32) ::server_id && - !replicate_same_server_id && - type_code != FORMAT_DESCRIPTION_EVENT) || - (rli->slave_skip_counter && - type_code != ROTATE_EVENT && type_code != STOP_EVENT && - type_code != START_EVENT_V3 && type_code!= FORMAT_DESCRIPTION_EVENT)) - { - DBUG_PRINT("info", ("event skipped")); - /* - We only skip the event here and do not increase the group log - position. In the event that we have to restart, this means - that we might have to skip the event again, but that is a - minor issue. - - If we were to increase the group log position when skipping an - event, it might be that we are restarting at the wrong - position and have events before that we should have executed, - so not increasing the group log position is a sure bet in this - case. - - In this way, we just step the group log position when we - *know* that we are at the end of a group. - */ - rli->inc_event_relay_log_pos(); - /* - Protect against common user error of setting the counter to 1 - instead of 2 while recovering from an insert which used auto_increment, - rand or user var. - */ - if (rli->slave_skip_counter && - !((type_code == INTVAR_EVENT || - type_code == RAND_EVENT || - type_code == USER_VAR_EVENT) && - rli->slave_skip_counter == 1) && - /* - The events from ourselves which have something to do with the relay - log itself must be skipped, true, but they mustn't decrement - rli->slave_skip_counter, because the user is supposed to not see - these events (they are not in the master's binlog) and if we - decremented, START SLAVE would for example decrement when it sees - the Rotate, so the event which the user probably wanted to skip - would not be skipped. - */ - !(ev->server_id == (uint32) ::server_id && - (type_code == ROTATE_EVENT || type_code == STOP_EVENT || - type_code == START_EVENT_V3 || type_code == FORMAT_DESCRIPTION_EVENT))) - --rli->slave_skip_counter; - pthread_mutex_unlock(&rli->data_lock); - delete ev; - DBUG_RETURN(0); // avoid infinite update loops - } - pthread_mutex_unlock(&rli->data_lock); + + /* + Execute the event to change the database and update the binary + log coordinates, but first we set some data that is needed for + the thread. + + The event will be executed unless it is supposed to be skipped. + + Queries originating from this server must be skipped. Low-level + events (Format_description_log_event, Rotate_log_event, + Stop_log_event) from this server must also be skipped. But for + those we don't want to modify 'group_master_log_pos', because + these events did not exist on the master. + Format_description_log_event is not completely skipped. + + Skip queries specified by the user in 'slave_skip_counter'. We + can't however skip events that has something to do with the log + files themselves. + + Filtering on own server id is extremely important, to ignore + execution of events created by the creation/rotation of the relay + log (remember that now the relay log starts with its Format_desc, + has a Rotate etc). + */ thd->server_id = ev->server_id; // use the original server id for logging thd->set_time(); // time the query @@ -1796,19 +1769,69 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) if (!ev->when) ev->when = time(NULL); ev->thd = thd; // because up to this point, ev->thd == 0 - DBUG_PRINT("info", ("thd->options={ %s%s}", - FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT), - FLAGSTR(thd->options, OPTION_BEGIN))); - exec_res = ev->exec_event(rli); - DBUG_PRINT("info", ("exec_event result: %d", exec_res)); - DBUG_ASSERT(rli->sql_thd==thd); + int reason= ev->shall_skip(rli); + if (reason == Log_event::EVENT_SKIP_COUNT) + --rli->slave_skip_counter; + pthread_mutex_unlock(&rli->data_lock); + if (reason == Log_event::EVENT_SKIP_NOT) + exec_res= ev->apply_event(rli); +#ifndef DBUG_OFF + else + { + /* + This only prints information to the debug trace. + + TODO: Print an informational message to the error log? + */ + static const char *const explain[] = { + "event was not skipped", // EVENT_SKIP_NOT, + "event originated from this server", // EVENT_SKIP_IGNORE, + "event skip counter was non-zero" // EVENT_SKIP_COUNT + }; + DBUG_PRINT("info", ("%s was skipped because %s", + ev->get_type_str(), explain[reason])); + } +#endif + + DBUG_PRINT("info", ("apply_event error = %d", exec_res)); + if (exec_res == 0) + { + int error= ev->update_pos(rli); + char buf[22]; + DBUG_PRINT("info", ("update_pos error = %d", error)); + DBUG_PRINT("info", ("group %s %s", + llstr(rli->group_relay_log_pos, buf), + rli->group_relay_log_name)); + DBUG_PRINT("info", ("event %s %s", + llstr(rli->event_relay_log_pos, buf), + rli->event_relay_log_name)); + /* + The update should not fail, so print an error message and + return an error code. + + TODO: Replace this with a decent error message when merged + with BUG#24954 (which adds several new error message). + */ + if (error) + { + slave_print_msg(ERROR_LEVEL, rli, ER_UNKNOWN_ERROR, + "It was not possible to update the positions" + " of the relay log information: the slave may" + " be in an inconsistent state." + " Stopped in %s position %s", + rli->group_relay_log_name, + llstr(rli->group_relay_log_pos, buf)); + DBUG_RETURN(1); + } + } + /* Format_description_log_event should not be deleted because it will be used to read info about the relay log's format; it will be deleted when the SQL thread does not need it, i.e. when this thread terminates. */ - if (ev->get_type_code() != FORMAT_DESCRIPTION_EVENT) + if (type_code != FORMAT_DESCRIPTION_EVENT) { DBUG_PRINT("info", ("Deleting the event after it has been executed")); delete ev; @@ -2369,13 +2392,17 @@ Slave SQL thread aborted. Can't execute init_slave query"); THD_CHECK_SENTRY(thd); if (exec_relay_log_event(thd,rli)) { + DBUG_PRINT("info", ("exec_relay_log_event() failed")); // do not scare the user if SQL thread was simply killed or stopped if (!sql_slave_killed(thd,rli)) { /* - retrieve as much info as possible from the thd and, error codes and warnings - and print this to the error log as to allow the user to locate the error + retrieve as much info as possible from the thd and, error + codes and warnings and print this to the error log as to + allow the user to locate the error */ + DBUG_PRINT("info", ("thd->net.last_errno=%d; rli->last_slave_errno=%d", + thd->net.last_errno, rli->last_slave_errno)); if (thd->net.last_errno != 0) { if (rli->last_slave_errno == 0) @@ -2396,10 +2423,25 @@ Slave SQL thread aborted. Can't execute init_slave query"); /* Print any warnings issued */ List_iterator_fast<MYSQL_ERROR> it(thd->warn_list); MYSQL_ERROR *err; + /* + Added controlled slave thread cancel for replication + of user-defined variables. + */ + bool udf_error = false; while ((err= it++)) + { + if (err->code == ER_CANT_OPEN_LIBRARY) + udf_error = true; sql_print_warning("Slave: %s Error_code: %d",err->msg, err->code); - - sql_print_error("\ + } + if (udf_error) + sql_print_error("Error loading user-defined library, slave SQL " + "thread aborted. Install the missing library, and restart the " + "slave SQL thread with \"SLAVE START\". We stopped at log '%s' " + "position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, + llbuff)); + else + sql_print_error("\ Error running query, slave SQL thread aborted. Fix the problem, and restart \ the slave SQL thread with \"SLAVE START\". We stopped at log \ '%s' position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, llbuff)); @@ -2702,6 +2744,7 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR)); DBUG_RETURN(1); } + pthread_mutex_lock(&mi->data_lock); ev->log_pos= mi->master_log_pos; /* 3.23 events don't contain log_pos */ switch (ev->get_type_code()) { @@ -2965,7 +3008,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) pthread_mutex_lock(log_lock); if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) && - !replicate_same_server_id) + !mi->rli.replicate_same_server_id) { /* Do not write it to the relay log. diff --git a/sql/slave.h b/sql/slave.h index f21266bbee4..107b74c09dd 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -162,9 +162,9 @@ bool show_binlog_info(THD* thd); bool rpl_master_has_bug(RELAY_LOG_INFO *rli, uint bug_id); const char *print_slave_db_safe(const char *db); -int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int error_code); +int check_expected_error(THD* thd, RELAY_LOG_INFO const *rli, int error_code); void skip_load_data_infile(NET* net); -void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli, +void slave_print_msg(enum loglevel level, RELAY_LOG_INFO const *rli, int err_code, const char* msg, ...) ATTRIBUTE_FORMAT(printf, 4, 5); @@ -182,7 +182,7 @@ int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,ulonglong pos, int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset, const char** errmsg); void set_slave_thread_options(THD* thd); -void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli); +void set_slave_thread_default_charset(THD *thd, RELAY_LOG_INFO const *rli); void rotate_relay_log(MASTER_INFO* mi); pthread_handler_t handle_slave_io(void *arg); diff --git a/sql/sql_base.cc b/sql/sql_base.cc index f83df06bb31..54bedc46fb8 100644 --- a/sql/sql_base.cc +++ b/sql/sql_base.cc @@ -858,6 +858,7 @@ void free_io_cache(TABLE *table) DBUG_VOID_RETURN; } + /* Close all tables which aren't in use by any thread @@ -970,6 +971,71 @@ bool close_cached_tables(THD *thd, bool if_wait_for_refresh, /* + Close all tables which match specified connection string or + if specified string is NULL, then any table with a connection string. +*/ + +bool close_cached_connection_tables(THD *thd, bool if_wait_for_refresh, + LEX_STRING *connection, bool have_lock) +{ + uint idx; + TABLE_LIST tmp, *tables= NULL; + bool result= FALSE; + DBUG_ENTER("close_cached_connections"); + DBUG_ASSERT(thd); + + bzero(&tmp, sizeof(TABLE_LIST)); + + if (!have_lock) + VOID(pthread_mutex_lock(&LOCK_open)); + + for (idx= 0; idx < table_def_cache.records; idx++) + { + TABLE_SHARE *share= (TABLE_SHARE *) hash_element(&table_def_cache, idx); + + /* Ignore if table is not open or does not have a connect_string */ + if (!share->connect_string.length || !share->ref_count) + continue; + + /* Compare the connection string */ + if (connection && + (connection->length > share->connect_string.length || + (connection->length < share->connect_string.length && + (share->connect_string.str[connection->length] != '/' && + share->connect_string.str[connection->length] != '\\')) || + strncasecmp(connection->str, share->connect_string.str, + connection->length))) + continue; + + /* close_cached_tables() only uses these elements */ + tmp.db= share->db.str; + tmp.table_name= share->table_name.str; + tmp.next_local= tables; + + tables= (TABLE_LIST *) memdup_root(thd->mem_root, (char*)&tmp, + sizeof(TABLE_LIST)); + } + + if (tables) + result= close_cached_tables(thd, FALSE, tables, TRUE); + + if (!have_lock) + VOID(pthread_mutex_unlock(&LOCK_open)); + + if (if_wait_for_refresh) + { + pthread_mutex_lock(&thd->mysys_var->mutex); + thd->mysys_var->current_mutex= 0; + thd->mysys_var->current_cond= 0; + thd->proc_info=0; + pthread_mutex_unlock(&thd->mysys_var->mutex); + } + + DBUG_RETURN(result); +} + + +/* Mark all tables in the list which were used by current substatement as free for reuse. @@ -1328,10 +1394,10 @@ void close_temporary_tables(THD *thd) due to special characters in the names */ append_identifier(thd, &s_query, table->s->db.str, strlen(table->s->db.str)); - s_query.q_append('.'); + s_query.append('.'); append_identifier(thd, &s_query, table->s->table_name.str, strlen(table->s->table_name.str)); - s_query.q_append(','); + s_query.append(','); next= table->next; close_temporary(table, 1, 1); } @@ -2272,6 +2338,9 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root, table->insert_values= 0; table->fulltext_searched= 0; table->file->ft_handler= 0; + /* Catch wrong handling of the auto_increment_field_not_null. */ + DBUG_ASSERT(!table->auto_increment_field_not_null); + table->auto_increment_field_not_null= FALSE; if (table->timestamp_field) table->timestamp_field_type= table->timestamp_field->get_auto_set_type(); table->pos_in_table_list= table_list; @@ -3746,7 +3815,7 @@ TABLE *open_temporary_table(THD *thd, const char *path, const char *db, tmp_table->reginfo.lock_type= TL_WRITE; // Simulate locked share->tmp_table= (tmp_table->file->has_transactions() ? - TRANSACTIONAL_TMP_TABLE : TMP_TABLE); + TRANSACTIONAL_TMP_TABLE : NON_TRANSACTIONAL_TMP_TABLE); if (link_in_list) { @@ -5793,7 +5862,6 @@ bool setup_tables_and_check_access(THD *thd, TABLE_LIST *leaves_tmp= NULL; bool first_table= true; - thd->leaf_count= 0; if (setup_tables(thd, context, from_clause, tables, &leaves_tmp, select_insert)) return TRUE; @@ -5811,7 +5879,6 @@ bool setup_tables_and_check_access(THD *thd, return TRUE; } first_table= 0; - thd->leaf_count++; } return FALSE; } @@ -6195,6 +6262,11 @@ err_no_arena: values values to fill with ignore_errors TRUE if we should ignore errors + NOTE + fill_record() may set table->auto_increment_field_not_null and a + caller should make sure that it is reset after their last call to this + function. + RETURN FALSE OK TRUE error occured @@ -6207,27 +6279,52 @@ fill_record(THD * thd, List<Item> &fields, List<Item> &values, List_iterator_fast<Item> f(fields),v(values); Item *value, *fld; Item_field *field; + TABLE *table= 0; DBUG_ENTER("fill_record"); + /* + Reset the table->auto_increment_field_not_null as it is valid for + only one row. + */ + if (fields.elements) + { + /* + On INSERT or UPDATE fields are checked to be from the same table, + thus we safely can take table from the first field. + */ + fld= (Item_field*)f++; + if (!(field= fld->filed_for_view_update())) + { + my_error(ER_NONUPDATEABLE_COLUMN, MYF(0), fld->name); + goto err; + } + table= field->field->table; + table->auto_increment_field_not_null= FALSE; + f.rewind(); + } while ((fld= f++)) { if (!(field= fld->filed_for_view_update())) { my_error(ER_NONUPDATEABLE_COLUMN, MYF(0), fld->name); - DBUG_RETURN(TRUE); + goto err; } value=v++; Field *rfield= field->field; - TABLE *table= rfield->table; + table= rfield->table; if (rfield == table->next_number_field) table->auto_increment_field_not_null= TRUE; if ((value->save_in_field(rfield, 0) < 0) && !ignore_errors) { my_message(ER_UNKNOWN_ERROR, ER(ER_UNKNOWN_ERROR), MYF(0)); - DBUG_RETURN(TRUE); + goto err; } } DBUG_RETURN(thd->net.report_error); +err: + if (table) + table->auto_increment_field_not_null= FALSE; + DBUG_RETURN(TRUE); } @@ -6276,6 +6373,11 @@ fill_record_n_invoke_before_triggers(THD *thd, List<Item> &fields, values list of fields ignore_errors TRUE if we should ignore errors + NOTE + fill_record() may set table->auto_increment_field_not_null and a + caller should make sure that it is reset after their last call to this + function. + RETURN FALSE OK TRUE error occured @@ -6286,19 +6388,38 @@ fill_record(THD *thd, Field **ptr, List<Item> &values, bool ignore_errors) { List_iterator_fast<Item> v(values); Item *value; + TABLE *table= 0; DBUG_ENTER("fill_record"); Field *field; + /* + Reset the table->auto_increment_field_not_null as it is valid for + only one row. + */ + if (*ptr) + { + /* + On INSERT or UPDATE fields are checked to be from the same table, + thus we safely can take table from the first field. + */ + table= (*ptr)->table; + table->auto_increment_field_not_null= FALSE; + } while ((field = *ptr++)) { value=v++; - TABLE *table= field->table; + table= field->table; if (field == table->next_number_field) table->auto_increment_field_not_null= TRUE; if (value->save_in_field(field, 0) < 0) - DBUG_RETURN(TRUE); + goto err; } DBUG_RETURN(thd->net.report_error); + +err: + if (table) + table->auto_increment_field_not_null= FALSE; + DBUG_RETURN(TRUE); } diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc index b0a54bec664..6f7bbda96de 100644 --- a/sql/sql_binlog.cc +++ b/sql/sql_binlog.cc @@ -163,9 +163,17 @@ void mysql_client_binlog_statement(THD* thd) (ulong) uint4korr(bufptr+EVENT_LEN_OFFSET))); #endif ev->thd= thd; - if (IF_DBUG(int err= ) ev->exec_event(thd->rli_fake)) + /* + We go directly to the application phase, since we don't need + to check if the event shall be skipped or not. + + Neither do we have to update the log positions, since that is + not used at all: the rli_fake instance is used only for error + reporting. + */ + if (IF_DBUG(int err= ) ev->apply_event(thd->rli_fake)) { - DBUG_PRINT("error", ("exec_event() returned: %d", err)); + DBUG_PRINT("info", ("apply_event() returned: %d", err)); /* TODO: Maybe a better error message since the BINLOG statement now contains several events. diff --git a/sql/sql_class.cc b/sql/sql_class.cc index d8371bef446..cc38d63c9f9 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -679,11 +679,22 @@ bool THD::store_globals() void THD::cleanup_after_query() { + /* + Reset rand_used so that detection of calls to rand() will save random + seeds if needed by the slave. + + Do not reset rand_used if inside a stored function or trigger because + only the call to these operations is logged. Thus only the calling + statement needs to detect rand() calls made by its substatements. These + substatements must not set rand_used to 0 because it would remove the + detection of rand() by the calling statement. + */ if (!in_sub_stmt) /* stored functions and triggers are a special case */ { /* Forget those values, for next binlogger: */ stmt_depends_on_first_successful_insert_id_in_prev_stmt= 0; auto_inc_intervals_in_cur_stmt_for_binlog.empty(); + rand_used= 0; } if (first_successful_insert_id_in_cur_stmt > 0) { @@ -2542,30 +2553,111 @@ my_size_t THD::max_row_length_blob(TABLE *table, const byte *data) const } -my_size_t THD::pack_row(TABLE *table, MY_BITMAP const* cols, byte *row_data, - const byte *record) const +/* + Pack a record of data for a table into a format suitable for + transfer via the binary log. + + SYNOPSIS + THD::pack_row() + table Table describing the format of the record + cols Bitmap with a set bit for each column that should be + stored in the row + row_data Pointer to memory where row will be written + record Pointer to record that should be packed. It is assumed + that the pointer refers to either record[0] or + record[1], but no such check is made since the code does + not rely on that. + + DESCRIPTION + + The format for a row in transfer with N fields is the following: + + ceil(N/8) null bytes: + One null bit for every column *regardless of whether it can be + null or not*. This simplifies the decoding. Observe that the + number of null bits is equal to the number of set bits in the + 'cols' bitmap. The number of null bytes is the smallest number + of bytes necessary to store the null bits. + + Padding bits are 1. + + N packets: + Each field is stored in packed format. + + + RETURN VALUE + + The number of bytes written at 'row_data'. + */ +my_size_t +THD::pack_row(TABLE *table, MY_BITMAP const* cols, + byte *const row_data, const byte *record) const { Field **p_field= table->field, *field; - int n_null_bytes= table->s->null_bytes; - byte *ptr; - uint i; + int const null_byte_count= (bitmap_bits_set(cols) + 7) / 8; + byte *pack_ptr = row_data + null_byte_count; + byte *null_ptr = row_data; my_ptrdiff_t const rec_offset= record - table->record[0]; my_ptrdiff_t const def_offset= table->s->default_values - table->record[0]; - memcpy(row_data, record, n_null_bytes); - ptr= row_data+n_null_bytes; - for (i= 0 ; (field= *p_field) ; i++, p_field++) + /* + We write the null bits and the packed records using one pass + through all the fields. The null bytes are written little-endian, + i.e., the first fields are in the first byte. + */ + unsigned int null_bits= (1U << 8) - 1; + // Mask to mask out the correct but among the null bits + unsigned int null_mask= 1U; + for ( ; (field= *p_field) ; p_field++) { - if (bitmap_is_set(cols,i)) + DBUG_PRINT("debug", ("null_mask=%d; null_ptr=%p; row_data=%p; null_byte_count=%d", + null_mask, null_ptr, row_data, null_byte_count)); + if (bitmap_is_set(cols, p_field - table->field)) { - my_ptrdiff_t const offset= - field->is_null((uint) rec_offset) ? def_offset : rec_offset; - field->move_field_offset(offset); - ptr= (byte*)field->pack((char *) ptr, field->ptr); - field->move_field_offset(-offset); + my_ptrdiff_t offset; + if (field->is_null(rec_offset)) + { + offset= def_offset; + null_bits |= null_mask; + } + else + { + offset= rec_offset; + null_bits &= ~null_mask; + + /* + We only store the data of the field if it is non-null + */ + pack_ptr= (byte*)field->pack((char *) pack_ptr, field->ptr + offset); + } + + null_mask <<= 1; + if ((null_mask & 0xFF) == 0) + { + DBUG_ASSERT(null_ptr < row_data + null_byte_count); + null_mask = 1U; + *null_ptr++ = null_bits; + null_bits= (1U << 8) - 1; + } } } - return (static_cast<my_size_t>(ptr - row_data)); + + /* + Write the last (partial) byte, if there is one + */ + if ((null_mask & 0xFF) > 1) + { + DBUG_ASSERT(null_ptr < row_data + null_byte_count); + *null_ptr++ = null_bits; + } + + /* + The null pointer should now point to the first byte of the + packed data. If it doesn't, something is very wrong. + */ + DBUG_ASSERT(null_ptr == row_data + null_byte_count); + + return static_cast<my_size_t>(pack_ptr - row_data); } diff --git a/sql/sql_class.h b/sql/sql_class.h index 6b79917f772..de1296cede5 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -1270,7 +1270,7 @@ public: return first_successful_insert_id_in_prev_stmt; } /* - Used by Intvar_log_event::exec_event() and by "SET INSERT_ID=#" + Used by Intvar_log_event::do_apply_event() and by "SET INSERT_ID=#" (mysqlbinlog). We'll soon add a variant which can take many intervals in argument. */ @@ -1412,9 +1412,6 @@ public: partition_info *work_part_info; #endif - /* pass up the count of "leaf" tables in a JOIN out of setup_tables() */ - byte leaf_count; - THD(); ~THD(); diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc index 0ff5c4e5b50..d300edd6e18 100644 --- a/sql/sql_delete.cc +++ b/sql/sql_delete.cc @@ -54,6 +54,27 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds, if (mysql_prepare_delete(thd, table_list, &conds)) DBUG_RETURN(TRUE); + /* check ORDER BY even if it can be ignored */ + if (order && order->elements) + { + TABLE_LIST tables; + List<Item> fields; + List<Item> all_fields; + + bzero((char*) &tables,sizeof(tables)); + tables.table = table; + tables.alias = table_list->alias; + + if (select_lex->setup_ref_array(thd, order->elements) || + setup_order(thd, select_lex->ref_pointer_array, &tables, + fields, all_fields, (ORDER*) order->first)) + { + delete select; + free_underlaid_joins(thd, &thd->lex->select_lex); + DBUG_RETURN(TRUE); + } + } + const_cond= (!conds || conds->const_item()); safe_update=test(thd->options & OPTION_SAFE_UPDATES); if (safe_update && const_cond) @@ -155,23 +176,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds, { uint length= 0; SORT_FIELD *sortorder; - TABLE_LIST tables; - List<Item> fields; - List<Item> all_fields; ha_rows examined_rows; - - bzero((char*) &tables,sizeof(tables)); - tables.table = table; - tables.alias = table_list->alias; - - if (select_lex->setup_ref_array(thd, order->elements) || - setup_order(thd, select_lex->ref_pointer_array, &tables, - fields, all_fields, (ORDER*) order->first)) - { - delete select; - free_underlaid_joins(thd, &thd->lex->select_lex); - DBUG_RETURN(TRUE); - } if ((!select || table->quick_keys.is_clear_all()) && limit != HA_POS_ERROR) usable_index= get_index_for_order(table, (ORDER*)(order->first), limit); diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc index 5712fbceac8..89bd7958c86 100644 --- a/sql/sql_derived.cc +++ b/sql/sql_derived.cc @@ -179,7 +179,7 @@ exit: orig_table_list->table_name= table->s->table_name.str; orig_table_list->table_name_length= table->s->table_name.length; table->derived_select_number= first_select->select_number; - table->s->tmp_table= TMP_TABLE; + table->s->tmp_table= NON_TRANSACTIONAL_TMP_TABLE; #ifndef NO_EMBEDDED_ACCESS_CHECKS if (orig_table_list->referencing_view) table->grant= orig_table_list->grant; diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc index 773383c2c2d..c0e0203ed86 100644 --- a/sql/sql_insert.cc +++ b/sql/sql_insert.cc @@ -753,6 +753,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list, table->next_number_field->val_int() : 0)); table->next_number_field=0; thd->count_cuted_fields= CHECK_FIELD_IGNORE; + table->auto_increment_field_not_null= FALSE; if (duplic != DUP_ERROR || ignore) table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY); if (duplic == DUP_REPLACE && @@ -2466,14 +2467,12 @@ bool mysql_insert_select_prepare(THD *thd) DBUG_ASSERT(select_lex->leaf_tables != 0); lex->leaf_tables_insert= select_lex->leaf_tables; /* skip all leaf tables belonged to view where we are insert */ - for (first_select_leaf_table= select_lex->leaf_tables->next_leaf, - thd->leaf_count --; + for (first_select_leaf_table= select_lex->leaf_tables->next_leaf; first_select_leaf_table && first_select_leaf_table->belong_to_view && first_select_leaf_table->belong_to_view == lex->leaf_tables_insert->belong_to_view; - first_select_leaf_table= first_select_leaf_table->next_leaf, - thd->leaf_count --) + first_select_leaf_table= first_select_leaf_table->next_leaf) {} select_lex->leaf_tables= first_select_leaf_table; DBUG_RETURN(FALSE); @@ -2691,6 +2690,7 @@ select_insert::~select_insert() if (table) { table->next_number_field=0; + table->auto_increment_field_not_null= FALSE; table->file->ha_reset(); } thd->count_cuted_fields= CHECK_FIELD_IGNORE; diff --git a/sql/sql_list.h b/sql/sql_list.h index af30cbe0d6a..ba61a931e04 100644 --- a/sql/sql_list.h +++ b/sql/sql_list.h @@ -23,7 +23,7 @@ class Sql_alloc { public: - static void *operator new(size_t size) + static void *operator new(size_t size) throw () { return (void*) sql_alloc((uint) size); } @@ -31,9 +31,9 @@ public: { return (void*) sql_alloc((uint) size); } - static void *operator new[](size_t size, MEM_ROOT *mem_root) + static void *operator new[](size_t size, MEM_ROOT *mem_root) throw () { return (void*) alloc_root(mem_root, (uint) size); } - static void *operator new(size_t size, MEM_ROOT *mem_root) + static void *operator new(size_t size, MEM_ROOT *mem_root) throw () { return (void*) alloc_root(mem_root, (uint) size); } static void operator delete(void *ptr, size_t size) { TRASH(ptr, size); } static void operator delete(void *ptr, MEM_ROOT *mem_root) diff --git a/sql/sql_load.cc b/sql/sql_load.cc index cf356a4b336..7d2c2281bba 100644 --- a/sql/sql_load.cc +++ b/sql/sql_load.cc @@ -512,6 +512,7 @@ err: mysql_unlock_tables(thd, thd->lock); thd->lock=0; } + table->auto_increment_field_not_null= FALSE; thd->abort_on_warning= 0; DBUG_RETURN(error); } @@ -609,8 +610,6 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list, { uint length; byte save_chr; - if (field == table->next_number_field) - table->auto_increment_field_not_null= TRUE; if ((length=(uint) (read_info.row_end-pos)) > field->field_length) length=field->field_length; diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index 7e7e814a833..0c18d38e635 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -2595,6 +2595,36 @@ end_with_restore_list: break; } case SQLCOM_REPLACE: +#ifndef DBUG_OFF + if (mysql_bin_log.is_open()) + { + /* + Generate an incident log event before writing the real event + to the binary log. We put this event is before the statement + since that makes it simpler to check that the statement was + not executed on the slave (since incidents usually stop the + slave). + + Observe that any row events that are generated will be + generated before. + + This is only for testing purposes and will not be present in a + release build. + */ + + Incident incident= INCIDENT_NONE; + DBUG_PRINT("debug", ("Just before generate_incident()")); + DBUG_EXECUTE_IF("incident_database_resync_on_replace", + incident= INCIDENT_LOST_EVENTS;); + if (incident) + { + Incident_log_event ev(thd, incident); + mysql_bin_log.write(&ev); + mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE); + } + DBUG_PRINT("debug", ("Just after generate_incident()")); + } +#endif case SQLCOM_INSERT: { DBUG_ASSERT(first_table == all_tables && first_table != 0); @@ -3923,7 +3953,6 @@ create_sp_error: if (check_access(thd, DELETE_ACL, "mysql", 0, 1, 0, 0)) goto error; - /* Does NOT write to binlog */ if (!(res = mysql_drop_function(thd, &lex->spname->m_name))) { send_ok(thd); @@ -4274,6 +4303,10 @@ create_sp_error: int error; LEX *lex= thd->lex; DBUG_PRINT("info", ("case SQLCOM_CREATE_SERVER")); + + if (check_global_access(thd, SUPER_ACL)) + break; + if ((error= create_server(thd, &lex->server_options))) { DBUG_PRINT("info", ("problem creating server <%s>", @@ -4289,6 +4322,10 @@ create_sp_error: int error; LEX *lex= thd->lex; DBUG_PRINT("info", ("case SQLCOM_ALTER_SERVER")); + + if (check_global_access(thd, SUPER_ACL)) + break; + if ((error= alter_server(thd, &lex->server_options))) { DBUG_PRINT("info", ("problem altering server <%s>", @@ -4304,9 +4341,13 @@ create_sp_error: int err_code; LEX *lex= thd->lex; DBUG_PRINT("info", ("case SQLCOM_DROP_SERVER")); + + if (check_global_access(thd, SUPER_ACL)) + break; + if ((err_code= drop_server(thd, &lex->server_options))) { - if (! lex->drop_if_exists && err_code == ER_FOREIGN_SERVER_EXISTS) + if (! lex->drop_if_exists && err_code == ER_FOREIGN_SERVER_DOESNT_EXIST) { DBUG_PRINT("info", ("problem dropping server %s", lex->server_options.server_name)); @@ -5185,6 +5226,7 @@ void mysql_init_multi_delete(LEX *lex) lex->query_tables_last= &lex->query_tables; } + /* When you modify mysql_parse(), you may need to mofify mysql_test_parse_for_slave() in this same file. @@ -5197,6 +5239,7 @@ void mysql_parse(THD *thd, char *inBuf, uint length) DBUG_EXECUTE_IF("parser_debug", turn_parser_debug_on();); mysql_init_query(thd, (uchar*) inBuf, length); + if (query_cache_send_result_to_client(thd, inBuf, length) <= 0) { LEX *lex= thd->lex; diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 582df11c8c1..eb6f4efccc3 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -413,7 +413,12 @@ JOIN::prepare(Item ***rref_pointer_array, tables_list, &select_lex->leaf_tables, FALSE, SELECT_ACL, SELECT_ACL)) DBUG_RETURN(-1); - tables= thd->leaf_count; + + TABLE_LIST *table_ptr; + for (table_ptr= select_lex->leaf_tables; + table_ptr; + table_ptr= table_ptr->next_leaf) + tables++; if (setup_wild(thd, tables_list, fields_list, &all_fields, wild_num) || select_lex->setup_ref_array(thd, og_num) || @@ -7115,6 +7120,7 @@ static bool check_simple_equality(Item *left_item, Item *right_item, SYNOPSIS check_row_equality() + thd thread handle left_row left term of the row equality to be processed right_row right term of the row equality to be processed cond_equal multiple equalities that must hold together with the predicate @@ -7135,7 +7141,7 @@ static bool check_simple_equality(Item *left_item, Item *right_item, FALSE otherwise */ -static bool check_row_equality(Item *left_row, Item_row *right_row, +static bool check_row_equality(THD *thd, Item *left_row, Item_row *right_row, COND_EQUAL *cond_equal, List<Item>* eq_list) { uint n= left_row->cols(); @@ -7146,13 +7152,21 @@ static bool check_row_equality(Item *left_row, Item_row *right_row, Item *right_item= right_row->element_index(i); if (left_item->type() == Item::ROW_ITEM && right_item->type() == Item::ROW_ITEM) - is_converted= check_row_equality((Item_row *) left_item, - (Item_row *) right_item, - cond_equal, eq_list); - else + { + is_converted= check_row_equality(thd, + (Item_row *) left_item, + (Item_row *) right_item, + cond_equal, eq_list); + if (!is_converted) + thd->lex->current_select->cond_count++; + } + else + { is_converted= check_simple_equality(left_item, right_item, 0, cond_equal); - - if (!is_converted) + thd->lex->current_select->cond_count++; + } + + if (!is_converted) { Item_func_eq *eq_item; if (!(eq_item= new Item_func_eq(left_item, right_item))) @@ -7171,6 +7185,7 @@ static bool check_row_equality(Item *left_row, Item_row *right_row, SYNOPSIS check_equality() + thd thread handle item predicate to process cond_equal multiple equalities that must hold together with the predicate eq_list results of conversions of row equalities that are not simple @@ -7195,7 +7210,7 @@ static bool check_row_equality(Item *left_row, Item_row *right_row, or, if the procedure fails by a fatal error. */ -static bool check_equality(Item *item, COND_EQUAL *cond_equal, +static bool check_equality(THD *thd, Item *item, COND_EQUAL *cond_equal, List<Item> *eq_list) { if (item->type() == Item::FUNC_ITEM && @@ -7206,9 +7221,13 @@ static bool check_equality(Item *item, COND_EQUAL *cond_equal, if (left_item->type() == Item::ROW_ITEM && right_item->type() == Item::ROW_ITEM) - return check_row_equality((Item_row *) left_item, + { + thd->lex->current_select->cond_count--; + return check_row_equality(thd, + (Item_row *) left_item, (Item_row *) right_item, cond_equal, eq_list); + } else return check_simple_equality(left_item, right_item, item, cond_equal); } @@ -7221,6 +7240,7 @@ static bool check_equality(Item *item, COND_EQUAL *cond_equal, SYNOPSIS build_equal_items_for_cond() + thd thread handle cond condition(expression) where to make replacement inherited path to all inherited multiple equality items @@ -7283,7 +7303,7 @@ static bool check_equality(Item *item, COND_EQUAL *cond_equal, pointer to the transformed condition */ -static COND *build_equal_items_for_cond(COND *cond, +static COND *build_equal_items_for_cond(THD *thd, COND *cond, COND_EQUAL *inherited) { Item_equal *item_equal; @@ -7316,7 +7336,7 @@ static COND *build_equal_items_for_cond(COND *cond, structure here because it's restored before each re-execution of any prepared statement/stored procedure. */ - if (check_equality(item, &cond_equal, &eq_list)) + if (check_equality(thd, item, &cond_equal, &eq_list)) li.remove(); } @@ -7351,7 +7371,7 @@ static COND *build_equal_items_for_cond(COND *cond, while ((item= li++)) { Item *new_item; - if ((new_item = build_equal_items_for_cond(item, inherited))!= item) + if ((new_item= build_equal_items_for_cond(thd, item, inherited)) != item) { /* This replacement happens only for standalone equalities */ /* @@ -7381,7 +7401,7 @@ static COND *build_equal_items_for_cond(COND *cond, for WHERE a=b AND c=d AND (b=c OR d=5) b=c is replaced by =(a,b,c,d). */ - if (check_equality(cond, &cond_equal, &eq_list)) + if (check_equality(thd, cond, &cond_equal, &eq_list)) { int n= cond_equal.current_level.elements + eq_list.elements; if (n == 0) @@ -7444,7 +7464,7 @@ static COND *build_equal_items_for_cond(COND *cond, SYNOPSIS build_equal_items() - thd Thread handler + thd thread handle cond condition to build the multiple equalities for inherited path to all inherited multiple equality items join_list list of join tables to which the condition refers to @@ -7505,7 +7525,7 @@ static COND *build_equal_items(THD *thd, COND *cond, if (cond) { - cond= build_equal_items_for_cond(cond, inherited); + cond= build_equal_items_for_cond(thd, cond, inherited); cond->update_used_tables(); if (cond->type() == Item::COND_ITEM && ((Item_cond*) cond)->functype() == Item_func::COND_AND_FUNC) @@ -8977,20 +8997,19 @@ static Field *create_tmp_field_from_item(THD *thd, Item *item, TABLE *table, enum enum_field_types type; /* - DATE/TIME fields have STRING_RESULT result type. To preserve - type they needed to be handled separately. + DATE/TIME and GEOMETRY fields have STRING_RESULT result type. + To preserve type they needed to be handled separately. */ if ((type= item->field_type()) == MYSQL_TYPE_DATETIME || type == MYSQL_TYPE_TIME || type == MYSQL_TYPE_DATE || - type == MYSQL_TYPE_TIMESTAMP) + type == MYSQL_TYPE_TIMESTAMP || type == MYSQL_TYPE_GEOMETRY) new_field= item->tmp_table_field_from_field_type(table, 1); /* Make sure that the blob fits into a Field_varstring which has 2-byte lenght. */ else if (item->max_length/item->collation.collation->mbmaxlen > 255 && - item->max_length/item->collation.collation->mbmaxlen < UINT_MAX16 - && convert_blob_length) + convert_blob_length < UINT_MAX16 && convert_blob_length) new_field= new Field_varstring(convert_blob_length, maybe_null, item->name, table->s, item->collation.collation); @@ -9417,13 +9436,19 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields, { if (item->with_sum_func && type != Item::SUM_FUNC_ITEM) { - /* - Mark that the we have ignored an item that refers to a summary - function. We need to know this if someone is going to use - DISTINCT on the result. - */ - param->using_indirect_summary_function=1; - continue; + if (item->used_tables() & OUTER_REF_TABLE_BIT) + item->update_used_tables(); + if (type == Item::SUBSELECT_ITEM || + (item->used_tables() & ~OUTER_REF_TABLE_BIT)) + { + /* + Mark that the we have ignored an item that refers to a summary + function. We need to know this if someone is going to use + DISTINCT on the result. + */ + param->using_indirect_summary_function=1; + continue; + } } if (item->const_item() && (int) hidden_field_count <= 0) continue; // We don't have to store this @@ -9608,6 +9633,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields, share->default_values= table->record[1]+alloc_length; } copy_func[0]=0; // End marker + param->func_count= copy_func - param->items_to_copy; setup_tmp_table_column_bitmaps(table, bitmaps); @@ -13824,6 +13850,7 @@ count_field_types(TMP_TABLE_PARAM *param, List<Item> &fields, if (!sum_item->quick_group) param->quick_group=0; // UDF SUM function param->sum_func_count++; + param->func_count++; for (uint i=0 ; i < sum_item->arg_count ; i++) { diff --git a/sql/sql_servers.cc b/sql/sql_servers.cc index 7b3b71cdd9a..d21864e8572 100644 --- a/sql/sql_servers.cc +++ b/sql/sql_servers.cc @@ -16,6 +16,21 @@ /* The servers are saved in the system table "servers" + + Currently, when the user performs an ALTER SERVER or a DROP SERVER + operation, it will cause all open tables which refer to the named + server connection to be flushed. This may cause some undesirable + behaviour with regard to currently running transactions. It is + expected that the DBA knows what s/he is doing when s/he performs + the ALTER SERVER or DROP SERVER operation. + + TODO: + It is desirable for us to implement a callback mechanism instead where + callbacks can be registered for specific server protocols. The callback + will be fired when such a server name has been created/altered/dropped + or when statistics are to be gathered such as how many actual connections. + Storage engines etc will be able to make use of the callback so that + currently running transactions etc will not be disrupted. */ #include "mysql_priv.h" @@ -25,15 +40,43 @@ #include "sp_head.h" #include "sp.h" -static my_bool servers_load(THD *thd, TABLE_LIST *tables); -HASH servers_cache; -pthread_mutex_t servers_cache_mutex; // To init the hash -uint servers_cache_initialised=FALSE; -/* Version of server table. incremented by servers_load */ -static uint servers_version=0; +/* + We only use 1 mutex to guard the data structures - THR_LOCK_servers. + Read locked when only reading data and write-locked for all other access. +*/ + +static HASH servers_cache; static MEM_ROOT mem; static rw_lock_t THR_LOCK_servers; +static bool get_server_from_table_to_cache(TABLE *table); + +/* insert functions */ +static int insert_server(THD *thd, FOREIGN_SERVER *server_options); +static int insert_server_record(TABLE *table, FOREIGN_SERVER *server); +static int insert_server_record_into_cache(FOREIGN_SERVER *server); +static void prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, + FOREIGN_SERVER *server); +/* drop functions */ +static int delete_server_record(TABLE *table, + char *server_name, + int server_name_length); +static int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options); + +/* update functions */ +static void prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, + FOREIGN_SERVER *existing, + FOREIGN_SERVER *altered); +static int update_server(THD *thd, FOREIGN_SERVER *existing, + FOREIGN_SERVER *altered); +static int update_server_record(TABLE *table, FOREIGN_SERVER *server); +static int update_server_record_in_cache(FOREIGN_SERVER *existing, + FOREIGN_SERVER *altered); +/* utility functions */ +static void merge_server_struct(FOREIGN_SERVER *from, FOREIGN_SERVER *to); + + + static byte *servers_cache_get_key(FOREIGN_SERVER *server, uint *length, my_bool not_used __attribute__((unused))) { @@ -46,6 +89,7 @@ static byte *servers_cache_get_key(FOREIGN_SERVER *server, uint *length, DBUG_RETURN((byte*) server->server_name); } + /* Initialize structures responsible for servers used in federated server scheme information for them from the server @@ -65,35 +109,27 @@ static byte *servers_cache_get_key(FOREIGN_SERVER *server, uint *length, 1 Could not initialize servers */ -my_bool servers_init(bool dont_read_servers_table) +bool servers_init(bool dont_read_servers_table) { THD *thd; - my_bool return_val= 0; + bool return_val= FALSE; DBUG_ENTER("servers_init"); /* init the mutex */ - if (pthread_mutex_init(&servers_cache_mutex, MY_MUTEX_INIT_FAST)) - DBUG_RETURN(1); - if (my_rwlock_init(&THR_LOCK_servers, NULL)) - DBUG_RETURN(1); + DBUG_RETURN(TRUE); /* initialise our servers cache */ if (hash_init(&servers_cache, system_charset_info, 32, 0, 0, (hash_get_key) servers_cache_get_key, 0, 0)) { - return_val= 1; /* we failed, out of memory? */ + return_val= TRUE; /* we failed, out of memory? */ goto end; } /* Initialize the mem root for data */ init_alloc_root(&mem, ACL_ALLOC_BLOCK_SIZE, 0); - /* - at this point, the cache is initialised, let it be known - */ - servers_cache_initialised= TRUE; - if (dont_read_servers_table) goto end; @@ -101,7 +137,7 @@ my_bool servers_init(bool dont_read_servers_table) To be able to run this from boot, we allocate a temporary THD */ if (!(thd=new THD)) - DBUG_RETURN(1); + DBUG_RETURN(TRUE); thd->thread_stack= (char*) &thd; thd->store_globals(); /* @@ -131,19 +167,13 @@ end: TRUE Error */ -static my_bool servers_load(THD *thd, TABLE_LIST *tables) +static bool servers_load(THD *thd, TABLE_LIST *tables) { TABLE *table; READ_RECORD read_record_info; - my_bool return_val= TRUE; + bool return_val= TRUE; DBUG_ENTER("servers_load"); - if (!servers_cache_initialised) - DBUG_RETURN(0); - - /* need to figure out how to utilise this variable */ - servers_version++; /* servers updated */ - /* first, send all cached rows to sleep with the fishes, oblivion! I expect this crappy comment replaced */ free_root(&mem, MYF(MY_MARK_BLOCKS_FREE)); @@ -157,7 +187,7 @@ static my_bool servers_load(THD *thd, TABLE_LIST *tables) goto end; } - return_val=0; + return_val= FALSE; end: end_read_record(&read_record_info); @@ -184,10 +214,10 @@ end: TRUE Failure */ -my_bool servers_reload(THD *thd) +bool servers_reload(THD *thd) { TABLE_LIST tables[1]; - my_bool return_val= 1; + bool return_val= TRUE; DBUG_ENTER("servers_reload"); if (thd->locked_tables) @@ -197,10 +227,9 @@ my_bool servers_reload(THD *thd) close_thread_tables(thd); } - /* - To avoid deadlocks we should obtain table locks before - obtaining servers_cache->lock mutex. - */ + DBUG_PRINT("info", ("locking servers_cache")); + rw_wrlock(&THR_LOCK_servers); + bzero((char*) tables, sizeof(tables)); tables[0].alias= tables[0].table_name= (char*) "servers"; tables[0].db= (char*) "mysql"; @@ -213,12 +242,6 @@ my_bool servers_reload(THD *thd) goto end; } - DBUG_PRINT("info", ("locking servers_cache")); - VOID(pthread_mutex_lock(&servers_cache_mutex)); - - //old_servers_cache= servers_cache; - //old_mem=mem; - if ((return_val= servers_load(thd, tables))) { // Error. Revert to old list /* blast, for now, we have no servers, discuss later way to preserve */ @@ -227,14 +250,14 @@ my_bool servers_reload(THD *thd) servers_free(); } - DBUG_PRINT("info", ("unlocking servers_cache")); - VOID(pthread_mutex_unlock(&servers_cache_mutex)); - end: close_thread_tables(thd); + DBUG_PRINT("info", ("unlocking servers_cache")); + rw_unlock(&THR_LOCK_servers); DBUG_RETURN(return_val); } + /* Initialize structures responsible for servers used in federated server scheme information for them from the server @@ -261,7 +284,8 @@ end: 1 could not insert server struct into global servers cache */ -my_bool get_server_from_table_to_cache(TABLE *table) +static bool +get_server_from_table_to_cache(TABLE *table) { /* alloc a server struct */ char *ptr; @@ -309,68 +333,6 @@ my_bool get_server_from_table_to_cache(TABLE *table) DBUG_RETURN(FALSE); } -/* - SYNOPSIS - server_exists_in_table() - THD *thd - thread pointer - LEX_SERVER_OPTIONS *server_options - pointer to Lex->server_options - - NOTES - This function takes a LEX_SERVER_OPTIONS struct, which is very much the - same type of structure as a FOREIGN_SERVER, it contains the values parsed - in any one of the [CREATE|DELETE|DROP] SERVER statements. Using the - member "server_name", index_read_idx either founds the record and returns - 1, or doesn't find the record, and returns 0 - - RETURN VALUES - 0 record not found - 1 record found -*/ - -my_bool server_exists_in_table(THD *thd, LEX_SERVER_OPTIONS *server_options) -{ - int result= 1; - int error= 0; - TABLE_LIST tables; - TABLE *table; - DBUG_ENTER("server_exists"); - - bzero((char*) &tables, sizeof(tables)); - tables.db= (char*) "mysql"; - tables.alias= tables.table_name= (char*) "servers"; - - /* need to open before acquiring THR_LOCK_plugin or it will deadlock */ - if (! (table= open_ltable(thd, &tables, TL_WRITE))) - DBUG_RETURN(TRUE); - - table->use_all_columns(); - - rw_wrlock(&THR_LOCK_servers); - VOID(pthread_mutex_lock(&servers_cache_mutex)); - - /* set the field that's the PK to the value we're looking for */ - table->field[0]->store(server_options->server_name, - server_options->server_name_length, - system_charset_info); - - if ((error= table->file->index_read_idx(table->record[0], 0, - (byte *)table->field[0]->ptr, HA_WHOLE_KEY, - HA_READ_KEY_EXACT))) - { - if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE) - { - table->file->print_error(error, MYF(0)); - result= -1; - } - result= 0; - DBUG_PRINT("info",("record for server '%s' not found!", - server_options->server_name)); - } - - VOID(pthread_mutex_unlock(&servers_cache_mutex)); - rw_unlock(&THR_LOCK_servers); - DBUG_RETURN(result); -} /* SYNOPSIS @@ -382,15 +344,18 @@ my_bool server_exists_in_table(THD *thd, LEX_SERVER_OPTIONS *server_options) This function takes a server object that is has all members properly prepared, ready to be inserted both into the mysql.servers table and the servers cache. + + THR_LOCK_servers must be write locked. RETURN VALUES 0 - no error other - error code */ -int insert_server(THD *thd, FOREIGN_SERVER *server) +static int +insert_server(THD *thd, FOREIGN_SERVER *server) { - int error= 0; + int error= -1; TABLE_LIST tables; TABLE *table; @@ -402,13 +367,7 @@ int insert_server(THD *thd, FOREIGN_SERVER *server) /* need to open before acquiring THR_LOCK_plugin or it will deadlock */ if (! (table= open_ltable(thd, &tables, TL_WRITE))) - DBUG_RETURN(TRUE); - - /* lock mutex to make sure no changes happen */ - VOID(pthread_mutex_lock(&servers_cache_mutex)); - - /* lock table */ - rw_wrlock(&THR_LOCK_servers); + goto end; /* insert the server into the table */ if ((error= insert_server_record(table, server))) @@ -419,12 +378,10 @@ int insert_server(THD *thd, FOREIGN_SERVER *server) goto end; end: - /* unlock the table */ - rw_unlock(&THR_LOCK_servers); - VOID(pthread_mutex_unlock(&servers_cache_mutex)); DBUG_RETURN(error); } + /* SYNOPSIS int insert_server_record_into_cache() @@ -434,13 +391,16 @@ end: This function takes a FOREIGN_SERVER pointer to an allocated (root mem) and inserts it into the global servers cache + THR_LOCK_servers must be write locked. + RETURN VALUE 0 - no error >0 - error code */ -int insert_server_record_into_cache(FOREIGN_SERVER *server) +static int +insert_server_record_into_cache(FOREIGN_SERVER *server) { int error=0; DBUG_ENTER("insert_server_record_into_cache"); @@ -461,6 +421,7 @@ int insert_server_record_into_cache(FOREIGN_SERVER *server) DBUG_RETURN(error); } + /* SYNOPSIS store_server_fields() @@ -478,7 +439,8 @@ int insert_server_record_into_cache(FOREIGN_SERVER *server) */ -void store_server_fields(TABLE *table, FOREIGN_SERVER *server) +static void +store_server_fields(TABLE *table, FOREIGN_SERVER *server) { table->use_all_columns(); @@ -539,6 +501,7 @@ void store_server_fields(TABLE *table, FOREIGN_SERVER *server) */ +static int insert_server_record(TABLE *table, FOREIGN_SERVER *server) { int error; @@ -605,9 +568,11 @@ int insert_server_record(TABLE *table, FOREIGN_SERVER *server) int drop_server(THD *thd, LEX_SERVER_OPTIONS *server_options) { - int error= 0; + int error; TABLE_LIST tables; TABLE *table; + LEX_STRING name= { server_options->server_name, + server_options->server_name_length }; DBUG_ENTER("drop_server"); DBUG_PRINT("info", ("server name server->server_name %s", @@ -617,28 +582,35 @@ int drop_server(THD *thd, LEX_SERVER_OPTIONS *server_options) tables.db= (char*) "mysql"; tables.alias= tables.table_name= (char*) "servers"; - /* need to open before acquiring THR_LOCK_plugin or it will deadlock */ - if (! (table= open_ltable(thd, &tables, TL_WRITE))) - DBUG_RETURN(TRUE); - rw_wrlock(&THR_LOCK_servers); - VOID(pthread_mutex_lock(&servers_cache_mutex)); + /* hit the memory hit first */ + if ((error= delete_server_record_in_cache(server_options))) + goto end; - if ((error= delete_server_record(table, - server_options->server_name, - server_options->server_name_length))) + if (! (table= open_ltable(thd, &tables, TL_WRITE))) + { + error= my_errno; goto end; + } + error= delete_server_record(table, name.str, name.length); - if ((error= delete_server_record_in_cache(server_options))) - goto end; + /* close the servers table before we call closed_cached_connection_tables */ + close_thread_tables(thd); + + if (close_cached_connection_tables(thd, TRUE, &name)) + { + push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_UNKNOWN_ERROR, "Server connection in use"); + } end: - VOID(pthread_mutex_unlock(&servers_cache_mutex)); rw_unlock(&THR_LOCK_servers); DBUG_RETURN(error); } + + /* SYNOPSIS @@ -657,10 +629,10 @@ end: */ -int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) +static int +delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) { - - int error= 0; + int error= ER_FOREIGN_SERVER_DOESNT_EXIST; FOREIGN_SERVER *server; DBUG_ENTER("delete_server_record_in_cache"); @@ -676,7 +648,7 @@ int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) DBUG_PRINT("info", ("server_name %s length %d not found!", server_options->server_name, server_options->server_name_length)); - // what should be done if not found in the cache? + goto end; } /* We succeded in deletion of the server to the table, now delete @@ -686,14 +658,15 @@ int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) server->server_name, server->server_name_length)); - if (server) - VOID(hash_delete(&servers_cache, (byte*) server)); - - servers_version++; /* servers updated */ + VOID(hash_delete(&servers_cache, (byte*) server)); + + error= 0; +end: DBUG_RETURN(error); } + /* SYNOPSIS @@ -713,6 +686,8 @@ int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) table for the particular server via the call to update_server_record, and in the servers_cache via update_server_record_in_cache. + THR_LOCK_servers must be write locked. + RETURN VALUE 0 - no error >0 - error code @@ -721,7 +696,7 @@ int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options) int update_server(THD *thd, FOREIGN_SERVER *existing, FOREIGN_SERVER *altered) { - int error= 0; + int error; TABLE *table; TABLE_LIST tables; DBUG_ENTER("update_server"); @@ -731,19 +706,26 @@ int update_server(THD *thd, FOREIGN_SERVER *existing, FOREIGN_SERVER *altered) tables.alias= tables.table_name= (char*)"servers"; if (!(table= open_ltable(thd, &tables, TL_WRITE))) - DBUG_RETURN(1); + { + error= my_errno; + goto end; + } - rw_wrlock(&THR_LOCK_servers); if ((error= update_server_record(table, altered))) goto end; - update_server_record_in_cache(existing, altered); + error= update_server_record_in_cache(existing, altered); + + /* + Perform a reload so we don't have a 'hole' in our mem_root + */ + servers_load(thd, &tables); end: - rw_unlock(&THR_LOCK_servers); DBUG_RETURN(error); } + /* SYNOPSIS @@ -760,6 +742,8 @@ end: HASH, then the updated record inserted, in essence replacing the old record. + THR_LOCK_servers must be write locked. + RETURN VALUE 0 - no error 1 - error @@ -790,13 +774,13 @@ int update_server_record_in_cache(FOREIGN_SERVER *existing, { DBUG_PRINT("info", ("had a problem inserting server %s at %lx", altered->server_name, (long unsigned int) altered)); - error= 1; + error= ER_OUT_OF_RESOURCES; } - servers_version++; /* servers updated */ DBUG_RETURN(error); } + /* SYNOPSIS @@ -829,9 +813,9 @@ void merge_server_struct(FOREIGN_SERVER *from, FOREIGN_SERVER *to) to->password= strdup_root(&mem, from->password); if (to->port == -1) to->port= from->port; - if (!to->socket) + if (!to->socket && from->socket) to->socket= strdup_root(&mem, from->socket); - if (!to->scheme) + if (!to->scheme && from->scheme) to->scheme= strdup_root(&mem, from->scheme); if (!to->owner) to->owner= strdup_root(&mem, from->owner); @@ -839,6 +823,7 @@ void merge_server_struct(FOREIGN_SERVER *from, FOREIGN_SERVER *to) DBUG_VOID_RETURN; } + /* SYNOPSIS @@ -861,7 +846,9 @@ void merge_server_struct(FOREIGN_SERVER *from, FOREIGN_SERVER *to) */ -int update_server_record(TABLE *table, FOREIGN_SERVER *server) + +static int +update_server_record(TABLE *table, FOREIGN_SERVER *server) { int error=0; DBUG_ENTER("update_server_record"); @@ -876,10 +863,7 @@ int update_server_record(TABLE *table, FOREIGN_SERVER *server) HA_READ_KEY_EXACT))) { if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE) - { table->file->print_error(error, MYF(0)); - error= 1; - } DBUG_PRINT("info",("server not found!")); error= ER_FOREIGN_SERVER_DOESNT_EXIST; } @@ -899,6 +883,7 @@ end: DBUG_RETURN(error); } + /* SYNOPSIS @@ -914,11 +899,11 @@ end: */ -int delete_server_record(TABLE *table, - char *server_name, - int server_name_length) +static int +delete_server_record(TABLE *table, + char *server_name, int server_name_length) { - int error= 0; + int error; DBUG_ENTER("delete_server_record"); table->use_all_columns(); @@ -930,10 +915,7 @@ int delete_server_record(TABLE *table, HA_READ_KEY_EXACT))) { if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE) - { table->file->print_error(error, MYF(0)); - error= 1; - } DBUG_PRINT("info",("server not found!")); error= ER_FOREIGN_SERVER_DOESNT_EXIST; } @@ -962,28 +944,35 @@ int delete_server_record(TABLE *table, int create_server(THD *thd, LEX_SERVER_OPTIONS *server_options) { - int error; + int error= ER_FOREIGN_SERVER_EXISTS; FOREIGN_SERVER *server; DBUG_ENTER("create_server"); DBUG_PRINT("info", ("server_options->server_name %s", server_options->server_name)); + rw_wrlock(&THR_LOCK_servers); + + /* hit the memory first */ + if (hash_search(&servers_cache, (byte*) server_options->server_name, + server_options->server_name_length)) + goto end; + server= (FOREIGN_SERVER *)alloc_root(&mem, sizeof(FOREIGN_SERVER)); - if ((error= prepare_server_struct_for_insert(server_options, server))) - goto end; + prepare_server_struct_for_insert(server_options, server); - if ((error= insert_server(thd, server))) - goto end; + error= insert_server(thd, server); DBUG_PRINT("info", ("error returned %d", error)); end: + rw_unlock(&THR_LOCK_servers); DBUG_RETURN(error); } + /* SYNOPSIS @@ -1000,37 +989,44 @@ end: int alter_server(THD *thd, LEX_SERVER_OPTIONS *server_options) { - int error= 0; + int error= ER_FOREIGN_SERVER_DOESNT_EXIST; FOREIGN_SERVER *altered, *existing; + LEX_STRING name= { server_options->server_name, + server_options->server_name_length }; DBUG_ENTER("alter_server"); DBUG_PRINT("info", ("server_options->server_name %s", server_options->server_name)); + rw_wrlock(&THR_LOCK_servers); + + if (!(existing= (FOREIGN_SERVER *) hash_search(&servers_cache, + (byte*) name.str, + name.length))) + goto end; + altered= (FOREIGN_SERVER *)alloc_root(&mem, sizeof(FOREIGN_SERVER)); - VOID(pthread_mutex_lock(&servers_cache_mutex)); + prepare_server_struct_for_update(server_options, existing, altered); - if (!(existing= (FOREIGN_SERVER *) hash_search(&servers_cache, - (byte*) server_options->server_name, - server_options->server_name_length))) - { - error= ER_FOREIGN_SERVER_DOESNT_EXIST; - goto end; - } + error= update_server(thd, existing, altered); - if ((error= prepare_server_struct_for_update(server_options, existing, altered))) - goto end; + /* close the servers table before we call closed_cached_connection_tables */ + close_thread_tables(thd); - if ((error= update_server(thd, existing, altered))) - goto end; + if (close_cached_connection_tables(thd, FALSE, &name)) + { + push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, + ER_UNKNOWN_ERROR, "Server connection in use"); + } end: DBUG_PRINT("info", ("error returned %d", error)); - VOID(pthread_mutex_unlock(&servers_cache_mutex)); + rw_unlock(&THR_LOCK_servers); DBUG_RETURN(error); } + /* SYNOPSIS @@ -1041,19 +1037,17 @@ end: NOTES RETURN VALUE - 0 - no error + none */ -int prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, - FOREIGN_SERVER *server) +static void +prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, + FOREIGN_SERVER *server) { - int error; char *unset_ptr= (char*)""; DBUG_ENTER("prepare_server_struct"); - error= 0; - /* these two MUST be set */ server->server_name= strdup_root(&mem, server_options->server_name); server->server_name_length= server_options->server_name_length; @@ -1083,7 +1077,7 @@ int prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, server->owner= server_options->owner ? strdup_root(&mem, server_options->owner) : unset_ptr; - DBUG_RETURN(error); + DBUG_VOID_RETURN; } /* @@ -1099,13 +1093,12 @@ int prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, */ -int prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, - FOREIGN_SERVER *existing, - FOREIGN_SERVER *altered) +static void +prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, + FOREIGN_SERVER *existing, + FOREIGN_SERVER *altered) { - int error; DBUG_ENTER("prepare_server_struct_for_update"); - error= 0; altered->server_name= strdup_root(&mem, server_options->server_name); altered->server_name_length= server_options->server_name_length; @@ -1156,7 +1149,7 @@ int prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, (strcmp(server_options->owner, existing->owner))) ? strdup_root(&mem, server_options->owner) : 0; - DBUG_RETURN(error); + DBUG_VOID_RETURN; } /* @@ -1175,16 +1168,65 @@ int prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, void servers_free(bool end) { DBUG_ENTER("servers_free"); - if (!servers_cache_initialised) + if (!hash_inited(&servers_cache)) + DBUG_VOID_RETURN; + if (!end) + { + free_root(&mem, MYF(MY_MARK_BLOCKS_FREE)); + my_hash_reset(&servers_cache); DBUG_VOID_RETURN; - VOID(pthread_mutex_destroy(&servers_cache_mutex)); - servers_cache_initialised=0; + } + rwlock_destroy(&THR_LOCK_servers); free_root(&mem,MYF(0)); hash_free(&servers_cache); DBUG_VOID_RETURN; } +/* + SYNOPSIS + + clone_server(MEM_ROOT *mem_root, FOREIGN_SERVER *orig, FOREIGN_SERVER *buff) + + Create a clone of FOREIGN_SERVER. If the supplied mem_root is of + thd->mem_root then the copy is automatically disposed at end of statement. + + NOTES + + ARGS + MEM_ROOT pointer (strings are copied into this mem root) + FOREIGN_SERVER pointer (made a copy of) + FOREIGN_SERVER buffer (if not-NULL, this pointer is returned) + + RETURN VALUE + FOREIGN_SEVER pointer (copy of one supplied FOREIGN_SERVER) +*/ + +static FOREIGN_SERVER *clone_server(MEM_ROOT *mem, const FOREIGN_SERVER *server, + FOREIGN_SERVER *buffer) +{ + DBUG_ENTER("sql_server.cc:clone_server"); + + if (!buffer) + buffer= (FOREIGN_SERVER *) alloc_root(mem, sizeof(FOREIGN_SERVER)); + + buffer->server_name= strmake_root(mem, server->server_name, + server->server_name_length); + buffer->port= server->port; + buffer->server_name_length= server->server_name_length; + + /* TODO: We need to examine which of these can really be NULL */ + buffer->db= server->db ? strdup_root(mem, server->db) : NULL; + buffer->scheme= server->scheme ? strdup_root(mem, server->scheme) : NULL; + buffer->username= server->username? strdup_root(mem, server->username): NULL; + buffer->password= server->password? strdup_root(mem, server->password): NULL; + buffer->socket= server->socket ? strdup_root(mem, server->socket) : NULL; + buffer->owner= server->owner ? strdup_root(mem, server->owner) : NULL; + buffer->host= server->host ? strdup_root(mem, server->host) : NULL; + + DBUG_RETURN(buffer); +} + /* @@ -1199,11 +1241,11 @@ void servers_free(bool end) */ -FOREIGN_SERVER *get_server_by_name(const char *server_name) +FOREIGN_SERVER *get_server_by_name(MEM_ROOT *mem, const char *server_name, + FOREIGN_SERVER *buff) { - ulong error_num=0; uint server_name_length; - FOREIGN_SERVER *server= 0; + FOREIGN_SERVER *server; DBUG_ENTER("get_server_by_name"); DBUG_PRINT("info", ("server_name %s", server_name)); @@ -1212,12 +1254,11 @@ FOREIGN_SERVER *get_server_by_name(const char *server_name) if (! server_name || !strlen(server_name)) { DBUG_PRINT("info", ("server_name not defined!")); - error_num= 1; DBUG_RETURN((FOREIGN_SERVER *)NULL); } DBUG_PRINT("info", ("locking servers_cache")); - VOID(pthread_mutex_lock(&servers_cache_mutex)); + rw_rdlock(&THR_LOCK_servers); if (!(server= (FOREIGN_SERVER *) hash_search(&servers_cache, (byte*) server_name, server_name_length))) @@ -1226,8 +1267,12 @@ FOREIGN_SERVER *get_server_by_name(const char *server_name) server_name, server_name_length)); server= (FOREIGN_SERVER *) NULL; } + /* otherwise, make copy of server */ + else + server= clone_server(mem, server, buff); + DBUG_PRINT("info", ("unlocking servers_cache")); - VOID(pthread_mutex_unlock(&servers_cache_mutex)); + rw_unlock(&THR_LOCK_servers); DBUG_RETURN(server); } diff --git a/sql/sql_servers.h b/sql/sql_servers.h index 23b8cefd5bb..63c691893d1 100644 --- a/sql/sql_servers.h +++ b/sql/sql_servers.h @@ -25,40 +25,19 @@ typedef struct st_federated_server } FOREIGN_SERVER; /* cache handlers */ -my_bool servers_init(bool dont_read_server_table); -my_bool servers_reload(THD *thd); -my_bool get_server_from_table_to_cache(TABLE *table); +bool servers_init(bool dont_read_server_table); +bool servers_reload(THD *thd); void servers_free(bool end=0); /* insert functions */ int create_server(THD *thd, LEX_SERVER_OPTIONS *server_options); -int insert_server(THD *thd, FOREIGN_SERVER *server_options); -int insert_server_record(TABLE *table, FOREIGN_SERVER *server); -int insert_server_record_into_cache(FOREIGN_SERVER *server); -void store_server_fields_for_insert(TABLE *table, FOREIGN_SERVER *server); -void store_server_fields_for_insert(TABLE *table, - FOREIGN_SERVER *existing, - FOREIGN_SERVER *altered); -int prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options, - FOREIGN_SERVER *server); /* drop functions */ int drop_server(THD *thd, LEX_SERVER_OPTIONS *server_options); -int delete_server_record(TABLE *table, - char *server_name, - int server_name_length); -int delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options); /* update functions */ int alter_server(THD *thd, LEX_SERVER_OPTIONS *server_options); -int prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options, - FOREIGN_SERVER *existing, - FOREIGN_SERVER *altered); -int update_server(THD *thd, FOREIGN_SERVER *existing, FOREIGN_SERVER *altered); -int update_server_record(TABLE *table, FOREIGN_SERVER *server); -int update_server_record_in_cache(FOREIGN_SERVER *existing, - FOREIGN_SERVER *altered); -/* utility functions */ -void merge_server_struct(FOREIGN_SERVER *from, FOREIGN_SERVER *to); -FOREIGN_SERVER *get_server_by_name(const char *server_name); -my_bool server_exists_in_table(THD *thd, char *server_name); + +/* lookup functions */ +FOREIGN_SERVER *get_server_by_name(MEM_ROOT *mem, const char *server_name, + FOREIGN_SERVER *server_buffer); diff --git a/sql/sql_show.cc b/sql/sql_show.cc index 5558b02ad78..77285978c86 100644 --- a/sql/sql_show.cc +++ b/sql/sql_show.cc @@ -53,7 +53,8 @@ enum enum_i_s_events_fields ISE_CREATED, ISE_LAST_ALTERED, ISE_LAST_EXECUTED, - ISE_EVENT_COMMENT + ISE_EVENT_COMMENT, + ISE_ORIGINATOR }; @@ -4403,10 +4404,23 @@ copy_event_to_schema_table(THD *thd, TABLE *sch_table, TABLE *event_table) } /* status */ - if (et.status == Event_timed::ENABLED) - sch_table->field[ISE_STATUS]->store(STRING_WITH_LEN("ENABLED"), scs); - else - sch_table->field[ISE_STATUS]->store(STRING_WITH_LEN("DISABLED"), scs); + + switch (et.status) + { + case Event_timed::ENABLED: + sch_table->field[ISE_STATUS]->store(STRING_WITH_LEN("ENABLED"), scs); + break; + case Event_timed::SLAVESIDE_DISABLED: + sch_table->field[ISE_STATUS]->store(STRING_WITH_LEN("SLAVESIDE_DISABLED"), + scs); + break; + case Event_timed::DISABLED: + sch_table->field[ISE_STATUS]->store(STRING_WITH_LEN("DISABLED"), scs); + break; + default: + DBUG_ASSERT(0); + } + sch_table->field[ISE_ORIGINATOR]->store(et.originator, TRUE); /* on_completion */ if (et.on_completion == Event_timed::ON_COMPLETION_DROP) @@ -4945,9 +4959,7 @@ int mysql_schema_table(THD *thd, LEX *lex, TABLE_LIST *table_list) TABLE *table; DBUG_ENTER("mysql_schema_table"); if (!(table= table_list->schema_table->create_table(thd, table_list))) - { DBUG_RETURN(1); - } table->s->tmp_table= SYSTEM_TMP_TABLE; table->grant.privilege= SELECT_ACL; /* @@ -5445,13 +5457,14 @@ ST_FIELD_INFO events_fields_info[]= {"SQL_MODE", 65535, MYSQL_TYPE_STRING, 0, 0, 0}, {"STARTS", 0, MYSQL_TYPE_TIMESTAMP, 0, 1, "Starts"}, {"ENDS", 0, MYSQL_TYPE_TIMESTAMP, 0, 1, "Ends"}, - {"STATUS", 8, MYSQL_TYPE_STRING, 0, 0, "Status"}, + {"STATUS", 18, MYSQL_TYPE_STRING, 0, 0, "Status"}, {"ON_COMPLETION", 12, MYSQL_TYPE_STRING, 0, 0, 0}, {"CREATED", 0, MYSQL_TYPE_TIMESTAMP, 0, 0, 0}, {"LAST_ALTERED", 0, MYSQL_TYPE_TIMESTAMP, 0, 0, 0}, {"LAST_EXECUTED", 0, MYSQL_TYPE_TIMESTAMP, 0, 1, 0}, {"EVENT_COMMENT", NAME_LEN, MYSQL_TYPE_STRING, 0, 0, 0}, - {0, 0, MYSQL_TYPE_STRING, 0, 0, 0} + {"ORIGINATOR", 10, MYSQL_TYPE_LONG, 0, 0, "Originator"}, + {0, 0, MYSQL_TYPE_STRING, 0, 0, 0} }; diff --git a/sql/sql_table.cc b/sql/sql_table.cc index bb3f293941e..3eb47ebae6e 100644 --- a/sql/sql_table.cc +++ b/sql/sql_table.cc @@ -6838,7 +6838,9 @@ copy_data_between_tables(TABLE *from,TABLE *to, copy_ptr->do_copy(copy_ptr); } prev_insert_id= to->file->next_insert_id; - if ((error=to->file->ha_write_row((byte*) to->record[0]))) + error=to->file->write_row((byte*) to->record[0]); + to->auto_increment_field_not_null= FALSE; + if (error) { if (!ignore || to->file->is_fatal_error(error, HA_CHECK_DUP)) diff --git a/sql/sql_udf.cc b/sql/sql_udf.cc index 4c584bff72a..da5c1b0bc66 100644 --- a/sql/sql_udf.cc +++ b/sql/sql_udf.cc @@ -403,6 +403,13 @@ int mysql_create_function(THD *thd,udf_func *udf) DBUG_RETURN(1); } + /* + Turn off row binlogging of this statement and use statement-based + so that all supporting tables are updated for CREATE FUNCTION command. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + rw_wrlock(&THR_LOCK_udf); if ((hash_search(&udf_hash,(byte*) udf->name.str, udf->name.length))) { @@ -466,6 +473,15 @@ int mysql_create_function(THD *thd,udf_func *udf) goto err; } rw_unlock(&THR_LOCK_udf); + + /* Binlog the create function. */ + if (mysql_bin_log.is_open()) + { + thd->clear_error(); + thd->binlog_query(THD::MYSQL_QUERY_TYPE, + thd->query, thd->query_length, FALSE, FALSE); + } + DBUG_RETURN(0); err: @@ -484,11 +500,20 @@ int mysql_drop_function(THD *thd,const LEX_STRING *udf_name) char *exact_name_str; uint exact_name_len; DBUG_ENTER("mysql_drop_function"); + if (!initialized) { my_message(ER_OUT_OF_RESOURCES, ER(ER_OUT_OF_RESOURCES), MYF(0)); DBUG_RETURN(1); } + + /* + Turn off row binlogging of this statement and use statement-based + so that all supporting tables are updated for DROP FUNCTION command. + */ + if (thd->current_stmt_binlog_row_based) + thd->clear_current_stmt_binlog_row_based(); + rw_wrlock(&THR_LOCK_udf); if (!(udf=(udf_func*) hash_search(&udf_hash,(byte*) udf_name->str, (uint) udf_name->length))) @@ -524,6 +549,15 @@ int mysql_drop_function(THD *thd,const LEX_STRING *udf_name) close_thread_tables(thd); rw_unlock(&THR_LOCK_udf); + + /* Binlog the drop function. */ + if (mysql_bin_log.is_open()) + { + thd->clear_error(); + thd->binlog_query(THD::MYSQL_QUERY_TYPE, + thd->query, thd->query_length, FALSE, FALSE); + } + DBUG_RETURN(0); err: rw_unlock(&THR_LOCK_udf); diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy index b7894e9ce04..36b09224796 100644 --- a/sql/sql_yacc.yy +++ b/sql/sql_yacc.yy @@ -948,6 +948,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize); %token SIGNED_SYM %token SIMPLE_SYM /* SQL-2003-N */ %token SLAVE +%token SLAVESIDE_DISABLE_SYM %token SMALLINT /* SQL-2003-R */ %token SNAPSHOT_SYM %token SOCKET_SYM @@ -1770,12 +1771,17 @@ ev_schedule_time: EVERY_SYM expr interval opt_ev_status: /* empty */ { $$= 0; } | ENABLE_SYM { - Lex->event_parse_data->status= Event_parse_data::ENABLED; + Lex->event_parse_data->status= Event_basic::ENABLED; + $$= 1; + } + | DISABLE_SYM ON SLAVE + { + Lex->event_parse_data->status= Event_basic::SLAVESIDE_DISABLED; $$= 1; } | DISABLE_SYM { - Lex->event_parse_data->status= Event_parse_data::DISABLED; + Lex->event_parse_data->status= Event_basic::DISABLED; $$= 1; } ; @@ -1805,13 +1811,13 @@ ev_on_completion: ON COMPLETION_SYM PRESERVE_SYM { Lex->event_parse_data->on_completion= - Event_parse_data::ON_COMPLETION_PRESERVE; + Event_basic::ON_COMPLETION_PRESERVE; $$= 1; } | ON COMPLETION_SYM NOT_SYM PRESERVE_SYM { Lex->event_parse_data->on_completion= - Event_parse_data::ON_COMPLETION_DROP; + Event_basic::ON_COMPLETION_DROP; $$= 1; } ; @@ -9998,6 +10004,7 @@ keyword_sp: | SIMPLE_SYM {} | SHARE_SYM {} | SHUTDOWN {} + | SLAVESIDE_DISABLE_SYM {} | SNAPSHOT_SYM {} | SOUNDS_SYM {} | SQL_CACHE_SYM {} diff --git a/sql/table.h b/sql/table.h index aa053f207b3..94df89ef992 100644 --- a/sql/table.h +++ b/sql/table.h @@ -58,7 +58,7 @@ typedef struct st_grant_info enum tmp_table_type { - NO_TMP_TABLE, TMP_TABLE, TRANSACTIONAL_TMP_TABLE, + NO_TMP_TABLE, NON_TRANSACTIONAL_TMP_TABLE, TRANSACTIONAL_TMP_TABLE, INTERNAL_TMP_TABLE, SYSTEM_TMP_TABLE }; @@ -439,6 +439,11 @@ struct st_table { my_bool no_cache; /* To signal that we should reset query_id for tables and cols */ my_bool clear_query_id; + /* + To indicate that a non-null value of the auto_increment field + was provided by the user or retrieved from the current record. + Used only in the MODE_NO_AUTO_VALUE_ON_ZERO mode. + */ my_bool auto_increment_field_not_null; my_bool insert_or_update; /* Can be used by the handler */ my_bool alias_name_used; /* true if table_name is alias */ |