From d75d8631ed2d6af730931ea7079ec7e512e61796 Mon Sep 17 00:00:00 2001 From: Monty Date: Fri, 20 Jan 2017 15:33:28 +0200 Subject: [MDEV-10570] Add Flashback support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ==== Description ==== Flashback can rollback the instances/databases/tables to an old snapshot. It's implement on Server-Level by full image format binary logs (--binlog-row-image=FULL), so it supports all engines. Currently, it’s a feature inside mysqlbinlog tool (with --flashback arguments). Because the flashback binlog events will store in the memory, you should check if there is enough memory in your machine. ==== New Arguments to mysqlbinlog ==== --flashback (-B) It will let mysqlbinlog to work on FLASHBACK mode. ==== New Arguments to mysqld ==== --flashback Setup the server to use flashback. This enables binary log in row mode and will enable extra logging for DDL's needed by flashback feature ==== Example ==== I have a table "t" in database "test", we can compare the output with "--flashback" and without. #client/mysqlbinlog /data/mysqldata_10.0/binlog/mysql-bin.000001 -vv -d test -T t --start-datetime="2013-03-27 14:54:00" > /tmp/1.sql #client/mysqlbinlog /data/mysqldata_10.0/binlog/mysql-bin.000001 -vv -d test -T t --start-datetime="2013-03-27 14:54:00" -B > /tmp/2.sql Then, importing the output flashback file (/tmp/2.log), it can flashback your database/table to the special time (--start-datetime). And if you know the exact postion, "--start-postion" is also works, mysqlbinlog will output the flashback logs that can flashback to "--start-postion" position. ==== Implement ==== 1. As we know, if binlog_format is ROW (binlog-row-image=FULL in 10.1 and later), all columns value are store in the row event, so we can get the data before mis-operation. 2. Just do following things: 2.1 Change Event Type, INSERT->DELETE, DELETE->INSERT. For example: INSERT INTO t VALUES (...) ---> DELETE FROM t WHERE ... DELETE FROM t ... ---> INSERT INTO t VALUES (...) 2.2 For Update_Event, swapping the SET part and WHERE part. For example: UPDATE t SET cols1 = vals1 WHERE cols2 = vals2 ---> UPDATE t SET cols2 = vals2 WHERE cols1 = vals1 2.3 For Multi-Rows Event, reverse the rows sequence, from the last row to the first row. For example: DELETE FROM t WHERE id=1; DELETE FROM t WHERE id=2; ...; DELETE FROM t WHERE id=n; ---> DELETE FROM t WHERE id=n; ...; DELETE FROM t WHERE id=2; DELETE FROM t WHERE id=1; 2.4 Output those events from the last one to the first one which mis-operation happened. For example: --- sql/log_event.cc | 418 +++++++++++++++++++++++++++++++++++++++++++++++++------ sql/log_event.h | 76 +++++++++- sql/mysqld.cc | 18 ++- sql/mysqld.h | 1 + sql/sys_vars.cc | 7 +- 5 files changed, 473 insertions(+), 47 deletions(-) (limited to 'sql') diff --git a/sql/log_event.cc b/sql/log_event.cc index 0af348ab453..893781223fb 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -300,17 +300,34 @@ public: constructor, but it would be possible to create a subclass holding the IO_CACHE itself. */ - Write_on_release_cache(IO_CACHE *cache, FILE *file, flag_set flags = 0) - : m_cache(cache), m_file(file), m_flags(flags) + Write_on_release_cache(IO_CACHE *cache, FILE *file, flag_set flags = 0, Log_event *ev = NULL) + : m_cache(cache), m_file(file), m_flags(flags), m_ev(ev) { reinit_io_cache(m_cache, WRITE_CACHE, 0L, FALSE, TRUE); } ~Write_on_release_cache() { +#ifdef MYSQL_CLIENT + if(m_ev == NULL) + { + copy_event_cache_to_file_and_reinit(m_cache, m_file); + if (m_flags & FLUSH_F) + fflush(m_file); + } + else // if m_ev<>NULL, then storing the output in output_buf + { + LEX_STRING tmp_str; + if (copy_event_cache_to_string_and_reinit(m_cache, &tmp_str)) + exit(1); + m_ev->output_buf.append(tmp_str.str, tmp_str.length); + my_free(tmp_str.str); + } +#else /* MySQL_SERVER */ copy_event_cache_to_file_and_reinit(m_cache, m_file); if (m_flags & FLUSH_F) fflush(m_file); +#endif } /* @@ -340,6 +357,7 @@ private: IO_CACHE *m_cache; FILE *m_file; flag_set m_flags; + Log_event *m_ev; // Used for Flashback }; /* @@ -2760,7 +2778,7 @@ log_event_print_value(IO_CACHE *file, const uchar *ptr, d= (ulong) (i64 / 1000000); t= (ulong) (i64 % 1000000); - my_b_printf(file, "%04d-%02d-%02d %02d:%02d:%02d", + my_b_printf(file, "'%04d-%02d-%02d %02d:%02d:%02d'", (int) (d / 10000), (int) (d % 10000) / 100, (int) (d % 100), (int) (t / 10000), (int) (t % 10000) / 100, (int) t % 100); return 8; @@ -2994,22 +3012,30 @@ size_t Rows_log_event::print_verbose_one_row(IO_CACHE *file, table_def *td, PRINT_EVENT_INFO *print_event_info, MY_BITMAP *cols_bitmap, - const uchar *value, const uchar *prefix) + const uchar *value, const uchar *prefix, + const my_bool no_fill_output) { const uchar *value0= value; const uchar *null_bits= value; uint null_bit_index= 0; char typestr[64]= ""; - + +#ifdef WHEN_FLASHBACK_REVIEW_READY + /* Storing the review SQL */ + IO_CACHE *review_sql= &print_event_info->review_sql_cache; + LEX_STRING review_str; +#endif + /* Skip metadata bytes which gives the information about nullabity of master columns. Master writes one bit for each affected column. */ value+= (bitmap_bits_set(cols_bitmap) + 7) / 8; - - my_b_printf(file, "%s", prefix); - + + if (!no_fill_output) + my_b_printf(file, "%s", prefix); + for (size_t i= 0; i < td->size(); i ++) { size_t size; @@ -3018,47 +3044,226 @@ Rows_log_event::print_verbose_one_row(IO_CACHE *file, table_def *td, if (bitmap_is_set(cols_bitmap, i) == 0) continue; - - my_b_printf(file, "### @%d=", static_cast(i + 1)); + + if (!no_fill_output) + my_b_printf(file, "### @%d=", static_cast(i + 1)); + if (!is_null) { size_t fsize= td->calc_field_size((uint)i, (uchar*) value); if (value + fsize > m_rows_end) { - my_b_printf(file, "***Corrupted replication event was detected." - " Not printing the value***\n"); + if (!no_fill_output) + my_b_printf(file, "***Corrupted replication event was detected." + " Not printing the value***\n"); value+= fsize; return 0; } } - if (!(size= log_event_print_value(file,is_null? NULL: value, - td->type(i), td->field_metadata(i), - typestr, sizeof(typestr)))) + + if (!no_fill_output) + { + size= log_event_print_value(file,is_null? NULL: value, + td->type(i), td->field_metadata(i), + typestr, sizeof(typestr)); +#ifdef WHEN_FLASHBACK_REVIEW_READY + if (need_flashback_review) + { + String tmp_str, hex_str; + IO_CACHE tmp_cache; + + // Using a tmp IO_CACHE to get the value output + open_cached_file(&tmp_cache, NULL, NULL, 0, MYF(MY_WME | MY_NABP)); + size= log_event_print_value(&tmp_cache, is_null? NULL: value, + td->type(i), td->field_metadata(i), + typestr, sizeof(typestr)); + if (copy_event_cache_to_string_and_reinit(&tmp_cache, &review_str)) + exit(1); + close_cached_file(&tmp_cache); + + switch (td->type(i)) // Converting a string to HEX format + { + case MYSQL_TYPE_VARCHAR: + case MYSQL_TYPE_VAR_STRING: + case MYSQL_TYPE_STRING: + case MYSQL_TYPE_BLOB: + // Avoid write_pos changed to a new area + // tmp_str.free(); + tmp_str.append(review_str.str + 1, review_str.length - 2); // Removing quotation marks + if (hex_str.alloc(tmp_str.length()*2+1)) // If out of memory + { + fprintf(stderr, "\nError: Out of memory. " + "Could not print correct binlog event.\n"); + exit(1); + } + octet2hex((char*) hex_str.ptr(), tmp_str.ptr(), tmp_str.length()); + my_b_printf(review_sql, ", UNHEX('%s')", hex_str.ptr()); + break; + default: + tmp_str.free(); + tmp_str.append(review_str.str, review_str.length); + my_b_printf(review_sql, ", %s", tmp_str.ptr()); + break; + } + my_free(revieww_str.str); + } +#endif + } + else + { + IO_CACHE tmp_cache; + open_cached_file(&tmp_cache, NULL, NULL, 0, MYF(MY_WME | MY_NABP)); + size= log_event_print_value(&tmp_cache,is_null? NULL: value, + td->type(i), td->field_metadata(i), + typestr, sizeof(typestr)); + close_cached_file(&tmp_cache); + } + + if (!size) return 0; if (!is_null) value+= size; - if (print_event_info->verbose > 1) + if (print_event_info->verbose > 1 && !no_fill_output) { my_b_write(file, (uchar*)" /* ", 4); my_b_printf(file, "%s ", typestr); - + my_b_printf(file, "meta=%d nullable=%d is_null=%d ", td->field_metadata(i), td->maybe_null(i), is_null); my_b_write(file, (uchar*)"*/", 2); } - - my_b_write_byte(file, '\n'); - + + if (!no_fill_output) + my_b_write_byte(file, '\n'); + null_bit_index++; } return value - value0; } +/** + Exchange the SET part and WHERE part for the Update events. + Revert the operations order for the Write and Delete events. + And then revert the events order from the last one to the first one. + + @param[in] print_event_info PRINT_EVENT_INFO + @param[in] rows_buff Packed event buff +*/ + +void Rows_log_event::change_to_flashback_event(PRINT_EVENT_INFO *print_event_info, + uchar *rows_buff, Log_event_type ev_type) +{ + Table_map_log_event *map; + table_def *td; + DYNAMIC_ARRAY rows_arr; + uchar *swap_buff1, *swap_buff2; + uchar *rows_pos= rows_buff + m_rows_before_size; + + if (!(map= print_event_info->m_table_map.get_table(m_table_id)) || + !(td= map->create_table_def())) + return; + + /* If the write rows event contained no values for the AI */ + if (((get_general_type_code() == WRITE_ROWS_EVENT) && (m_rows_buf==m_rows_end))) + goto end; + + (void) my_init_dynamic_array(&rows_arr, sizeof(LEX_STRING), 8, 8, MYF(0)); + + for (uchar *value= m_rows_buf; value < m_rows_end; ) + { + uchar *start_pos= value; + size_t length1= 0; + if (!(length1= print_verbose_one_row(NULL, td, print_event_info, + &m_cols, value, + (const uchar*) "", TRUE))) + { + fprintf(stderr, "\nError row length: %zu\n", length1); + exit(1); + } + value+= length1; + + swap_buff1= (uchar *) my_malloc(length1, MYF(0)); + if (!swap_buff1) + { + fprintf(stderr, "\nError: Out of memory. " + "Could not exchange to flashback event.\n"); + exit(1); + } + memcpy(swap_buff1, start_pos, length1); + + // For Update_event, we have the second part + size_t length2= 0; + if (ev_type == UPDATE_ROWS_EVENT || + ev_type == UPDATE_ROWS_EVENT_V1) + { + if (!(length2= print_verbose_one_row(NULL, td, print_event_info, + &m_cols, value, + (const uchar*) "", TRUE))) + { + fprintf(stderr, "\nError row length: %zu\n", length2); + exit(1); + } + value+= length2; + + swap_buff2= (uchar *) my_malloc(length2, MYF(0)); + if (!swap_buff2) + { + fprintf(stderr, "\nError: Out of memory. " + "Could not exchange to flashback event.\n"); + exit(1); + } + memcpy(swap_buff2, start_pos + length1, length2); // WHERE part + } + + if (ev_type == UPDATE_ROWS_EVENT || + ev_type == UPDATE_ROWS_EVENT_V1) + { + /* Swap SET and WHERE part */ + memcpy(start_pos, swap_buff2, length2); + memcpy(start_pos + length2, swap_buff1, length1); + } + + /* Free tmp buffers */ + my_free(swap_buff1); + if (ev_type == UPDATE_ROWS_EVENT || + ev_type == UPDATE_ROWS_EVENT_V1) + my_free(swap_buff2); + + /* Copying one row into a buff, and pushing into the array */ + LEX_STRING one_row; + + one_row.length= length1 + length2; + one_row.str= (char *) my_malloc(one_row.length, MYF(0)); + memcpy(one_row.str, start_pos, one_row.length); + if (one_row.str == NULL || push_dynamic(&rows_arr, (uchar *) &one_row)) + { + fprintf(stderr, "\nError: Out of memory. " + "Could not push flashback event into array.\n"); + exit(1); + } + } + + /* Copying rows from the end to the begining into event */ + for (uint i= rows_arr.elements; i > 0; --i) + { + LEX_STRING *one_row= dynamic_element(&rows_arr, i - 1, LEX_STRING*); + + memcpy(rows_pos, (uchar *)one_row->str, one_row->length); + rows_pos+= one_row->length; + my_free(one_row->str); + } + delete_dynamic(&rows_arr); + +end: + delete td; +} + + /** Print a row event into IO cache in human readable form (in SQL format) @@ -3071,8 +3276,12 @@ void Rows_log_event::print_verbose(IO_CACHE *file, Table_map_log_event *map; table_def *td; const char *sql_command, *sql_clause1, *sql_clause2; + const char *sql_command_short __attribute__((unused)); Log_event_type general_type_code= get_general_type_code(); - +#ifdef WHEN_FLASHBACK_REVIEW_READY + IO_CACHE *review_sql= &print_event_info->review_sql_cache; +#endif + if (m_extra_row_data) { uint8 extra_data_len= m_extra_row_data[EXTRA_ROW_INFO_LEN_OFFSET]; @@ -3102,19 +3311,23 @@ void Rows_log_event::print_verbose(IO_CACHE *file, sql_command= "INSERT INTO"; sql_clause1= "### SET\n"; sql_clause2= NULL; + sql_command_short= "I"; break; case DELETE_ROWS_EVENT: sql_command= "DELETE FROM"; sql_clause1= "### WHERE\n"; sql_clause2= NULL; + sql_command_short= "D"; break; case UPDATE_ROWS_EVENT: sql_command= "UPDATE"; sql_clause1= "### WHERE\n"; sql_clause2= "### SET\n"; + sql_command_short= "U"; break; default: sql_command= sql_clause1= sql_clause2= NULL; + sql_command_short= ""; DBUG_ASSERT(0); /* Not possible */ } @@ -3140,6 +3353,13 @@ void Rows_log_event::print_verbose(IO_CACHE *file, my_b_printf(file, "### %s %`s.%`s\n", sql_command, map->get_db_name(), map->get_table_name()); + +#ifdef WHEN_FLASHBACK_REVIEW_READY + if (need_flashback_review) + my_b_printf(review_sql, "\nINSERT INTO `%s`.`%s` VALUES ('%s'", + map->get_review_dbname(), map->get_review_tablename(), sql_command_short); +#endif + /* Print the first image */ if (!(length= print_verbose_one_row(file, td, print_event_info, &m_cols, value, @@ -3156,6 +3376,17 @@ void Rows_log_event::print_verbose(IO_CACHE *file, goto end; value+= length; } +#ifdef WHEN_FLASHBACK_REVIEW_READY + else + { + if (need_flashback_review) + for (size_t i= 0; i < td->size(); i ++) + my_b_printf(review_sql, ", NULL"); + } + + if (need_flashback_review) + my_b_printf(review_sql, ")%s\n", print_event_info->delimiter); +#endif } end: @@ -3171,7 +3402,7 @@ void Log_event::print_base64(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info, bool more) { - const uchar *ptr= (const uchar *)temp_buf; + uchar *ptr= (uchar *)temp_buf; uint32 size= uint4korr(ptr + EVENT_LEN_OFFSET); DBUG_ENTER("Log_event::print_base64"); @@ -3183,6 +3414,51 @@ void Log_event::print_base64(IO_CACHE* file, DBUG_VOID_RETURN; } + if (is_flashback) + { + uint tmp_size= size; + Rows_log_event *ev= NULL; + Log_event_type ev_type = (enum Log_event_type) ptr[EVENT_TYPE_OFFSET]; + if (checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF && + checksum_alg != BINLOG_CHECKSUM_ALG_OFF) + tmp_size-= BINLOG_CHECKSUM_LEN; // checksum is displayed through the header + switch (ev_type) { + case WRITE_ROWS_EVENT: + ptr[EVENT_TYPE_OFFSET]= DELETE_ROWS_EVENT; + ev= new Delete_rows_log_event((const char*) ptr, tmp_size, + glob_description_event); + ev->change_to_flashback_event(print_event_info, ptr, ev_type); + break; + case WRITE_ROWS_EVENT_V1: + ptr[EVENT_TYPE_OFFSET]= DELETE_ROWS_EVENT_V1; + ev= new Delete_rows_log_event((const char*) ptr, tmp_size, + glob_description_event); + ev->change_to_flashback_event(print_event_info, ptr, ev_type); + break; + case DELETE_ROWS_EVENT: + ptr[EVENT_TYPE_OFFSET]= WRITE_ROWS_EVENT; + ev= new Write_rows_log_event((const char*) ptr, tmp_size, + glob_description_event); + ev->change_to_flashback_event(print_event_info, ptr, ev_type); + break; + case DELETE_ROWS_EVENT_V1: + ptr[EVENT_TYPE_OFFSET]= WRITE_ROWS_EVENT_V1; + ev= new Write_rows_log_event((const char*) ptr, tmp_size, + glob_description_event); + ev->change_to_flashback_event(print_event_info, ptr, ev_type); + break; + case UPDATE_ROWS_EVENT: + case UPDATE_ROWS_EVENT_V1: + ev= new Update_rows_log_event((const char*) ptr, tmp_size, + glob_description_event); + ev->change_to_flashback_event(print_event_info, ptr, ev_type); + break; + default: + break; + } + delete ev; + } + if (my_base64_encode(ptr, (size_t) size, tmp_str)) { DBUG_ASSERT(0); @@ -3198,8 +3474,12 @@ void Log_event::print_base64(IO_CACHE* file, if (!more) my_b_printf(file, "'%s\n", print_event_info->delimiter); } - + +#ifdef WHEN_FLASHBACK_REVIEW_READY + if (print_event_info->verbose || need_flashback_review) +#else if (print_event_info->verbose) +#endif { Rows_log_event *ev= NULL; Log_event_type et= (Log_event_type) ptr[EVENT_TYPE_OFFSET]; @@ -3207,7 +3487,7 @@ void Log_event::print_base64(IO_CACHE* file, if (checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF && checksum_alg != BINLOG_CHECKSUM_ALG_OFF) size-= BINLOG_CHECKSUM_LEN; // checksum is displayed through the header - + switch (et) { case TABLE_MAP_EVENT: @@ -3215,6 +3495,13 @@ void Log_event::print_base64(IO_CACHE* file, Table_map_log_event *map; map= new Table_map_log_event((const char*) ptr, size, glob_description_event); +#ifdef WHEN_FLASHBACK_REVIEW_READY + if (need_flashback_review) + { + map->set_review_dbname(m_review_dbname.ptr()); + map->set_review_tablename(m_review_tablename.ptr()); + } +#endif print_event_info->m_table_map.set_table(map->get_table_id(), map); break; } @@ -3263,14 +3550,27 @@ void Log_event::print_base64(IO_CACHE* file, default: break; } - + if (ev) { +#ifdef WHEN_FLASHBACK_REVIEW_READY + ev->need_flashback_review= need_flashback_review; + if (print_event_info->verbose) + ev->print_verbose(file, print_event_info); + else + { + IO_CACHE tmp_cache; + open_cached_file(&tmp_cache, NULL, NULL, 0, MYF(MY_WME | MY_NABP)); + ev->print_verbose(&tmp_cache, print_event_info); + close_cached_file(&tmp_cache); + } +#else ev->print_verbose(file, print_event_info); +#endif delete ev; } } - + my_free(tmp_str); DBUG_VOID_RETURN; } @@ -4631,7 +4931,7 @@ void Query_log_event::print_query_header(IO_CACHE* file, void Query_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) { - Write_on_release_cache cache(&print_event_info->head_cache, file); + Write_on_release_cache cache(&print_event_info->head_cache, file, 0, this); /** reduce the size of io cache so that the write function is called @@ -4640,8 +4940,24 @@ void Query_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) DBUG_EXECUTE_IF ("simulate_file_write_error", {(&cache)->write_pos= (&cache)->write_end- 500;}); print_query_header(&cache, print_event_info); - my_b_write(&cache, (uchar*) query, q_len); - my_b_printf(&cache, "\n%s\n", print_event_info->delimiter); + if (!is_flashback) + { + my_b_write(&cache, (uchar*) query, q_len); + my_b_printf(&cache, "\n%s\n", print_event_info->delimiter); + } + else // is_flashback == 1 + { + if (strcmp("BEGIN", query) == 0) + { + my_b_write(&cache, (uchar*) "COMMIT", 6); + my_b_printf(&cache, "\n%s\n", print_event_info->delimiter); + } + else if (strcmp("COMMIT", query) == 0) + { + my_b_write(&cache, (uchar*) "BEGIN", 5); + my_b_printf(&cache, "\n%s\n", print_event_info->delimiter); + } + } } #endif /* MYSQL_CLIENT */ @@ -7351,11 +7667,11 @@ void Gtid_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info) { Write_on_release_cache cache(&print_event_info->head_cache, file, - Write_on_release_cache::FLUSH_F); + Write_on_release_cache::FLUSH_F, this); char buf[21]; char buf2[21]; - if (!print_event_info->short_form) + if (!print_event_info->short_form & !is_flashback) { print_header(&cache, print_event_info, FALSE); longlong10_to_str(seq_no, buf, 10); @@ -7401,11 +7717,12 @@ Gtid_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info) print_event_info->server_id_printed= true; } - my_b_printf(&cache, "/*!100001 SET @@session.gtid_seq_no=%s*/%s\n", - buf, print_event_info->delimiter); + if (!is_flashback) + my_b_printf(&cache, "/*!100001 SET @@session.gtid_seq_no=%s*/%s\n", + buf, print_event_info->delimiter); } if (!(flags2 & FL_STANDALONE)) - my_b_printf(&cache, "BEGIN\n%s\n", print_event_info->delimiter); + my_b_printf(&cache, is_flashback ? "COMMIT\n%s\n" : "BEGIN\n%s\n", print_event_info->delimiter); } #endif /* MYSQL_SERVER */ @@ -8047,7 +8364,7 @@ bool Xid_log_event::write() void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) { Write_on_release_cache cache(&print_event_info->head_cache, file, - Write_on_release_cache::FLUSH_F); + Write_on_release_cache::FLUSH_F, this); if (!print_event_info->short_form) { @@ -8057,7 +8374,7 @@ void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) print_header(&cache, print_event_info, FALSE); my_b_printf(&cache, "\tXid = %s\n", buf); } - my_b_printf(&cache, "COMMIT%s\n", print_event_info->delimiter); + my_b_printf(&cache, is_flashback ? "BEGIN%s\n" : "COMMIT%s\n", print_event_info->delimiter); } #endif /* MYSQL_CLIENT */ @@ -8706,7 +9023,7 @@ void Unknown_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info void Stop_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info) { Write_on_release_cache cache(&print_event_info->head_cache, file, - Write_on_release_cache::FLUSH_F); + Write_on_release_cache::FLUSH_F, this); if (print_event_info->short_form) return; @@ -10042,6 +10359,7 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len, m_rows_end= m_rows_buf + data_size; m_rows_cur= m_rows_end; memcpy(m_rows_buf, ptr_rows_data, data_size); + m_rows_before_size= ptr_rows_data - (const uchar *) buf; // Get the size that before SET part } else m_cols.bitmap= 0; // to not free it @@ -10958,6 +11276,10 @@ void Rows_log_event::print_helper(FILE *file, { IO_CACHE *const head= &print_event_info->head_cache; IO_CACHE *const body= &print_event_info->body_cache; +#ifdef WHEN_FLASHBACK_REVIEW_READY + IO_CACHE *const sql= &print_event_info->review_sql_cache; +#endif + if (!print_event_info->short_form) { bool const last_stmt_event= get_flags(STMT_END_F); @@ -10970,8 +11292,19 @@ void Rows_log_event::print_helper(FILE *file, if (get_flags(STMT_END_F)) { - copy_event_cache_to_file_and_reinit(head, file); - copy_event_cache_to_file_and_reinit(body, file); + reinit_io_cache(head, READ_CACHE, 0L, FALSE, FALSE); + output_buf.append(head, head->end_of_file); + reinit_io_cache(head, WRITE_CACHE, 0, FALSE, TRUE); + + reinit_io_cache(body, READ_CACHE, 0L, FALSE, FALSE); + output_buf.append(body, body->end_of_file); + reinit_io_cache(body, WRITE_CACHE, 0, FALSE, TRUE); + +#ifdef WHEN_FLASHBACK_REVIEW_READY + reinit_io_cache(sql, READ_CACHE, 0L, FALSE, FALSE); + output_buf.append(sql, sql->end_of_file); + reinit_io_cache(sql, WRITE_CACHE, 0, FALSE, TRUE); +#endif } } #endif @@ -12374,7 +12707,7 @@ void Write_rows_log_event::print(FILE *file, PRINT_EVENT_INFO* print_event_info) { DBUG_EXECUTE_IF("simulate_cache_read_error", {DBUG_SET("+d,simulate_my_b_fill_error");}); - Rows_log_event::print_helper(file, print_event_info, "Write_rows"); + Rows_log_event::print_helper(file, print_event_info, is_flashback ? "Delete_rows" : "Write_rows"); } void Write_rows_compressed_log_event::print(FILE *file, @@ -13048,7 +13381,7 @@ int Delete_rows_log_event::do_exec_row(rpl_group_info *rgi) void Delete_rows_log_event::print(FILE *file, PRINT_EVENT_INFO* print_event_info) { - Rows_log_event::print_helper(file, print_event_info, "Delete_rows"); + Rows_log_event::print_helper(file, print_event_info, is_flashback ? "Write_rows" : "Delete_rows"); } void Delete_rows_compressed_log_event::print(FILE *file, @@ -13600,6 +13933,9 @@ st_print_event_info::st_print_event_info() myf const flags = MYF(MY_WME | MY_NABP); open_cached_file(&head_cache, NULL, NULL, 0, flags); open_cached_file(&body_cache, NULL, NULL, 0, flags); +#ifdef WHEN_FLASHBACK_REVIEW_READY + open_cached_file(&review_sql_cache, NULL, NULL, 0, flags); +#endif } #endif diff --git a/sql/log_event.h b/sql/log_event.h index 7b8704636af..c2829a9bb10 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -41,6 +41,7 @@ #include "rpl_utility.h" #include "hash.h" #include "rpl_tblmap.h" +#include "sql_string.h" #endif #ifdef MYSQL_SERVER @@ -52,7 +53,9 @@ #include "rpl_gtid.h" /* Forward declarations */ +#ifndef MYSQL_CLIENT class String; +#endif #define PREFIX_SQL_LOAD "SQL_LOAD-" #define LONG_FIND_ROW_THRESHOLD 60 /* seconds */ @@ -845,9 +848,16 @@ typedef struct st_print_event_info ~st_print_event_info() { close_cached_file(&head_cache); close_cached_file(&body_cache); +#ifdef WHEN_FLASHBACK_REVIEW_READY + close_cached_file(&review_sql_cache); +#endif } bool init_ok() /* tells if construction was successful */ - { return my_b_inited(&head_cache) && my_b_inited(&body_cache); } + { return my_b_inited(&head_cache) && my_b_inited(&body_cache) +#ifdef WHEN_FLASHBACK_REVIEW_READY + && my_b_inited(&review_sql_cache) +#endif + ; } /* Settings on how to print the events */ @@ -875,6 +885,10 @@ typedef struct st_print_event_info */ IO_CACHE head_cache; IO_CACHE body_cache; +#ifdef WHEN_FLASHBACK_REVIEW_READY + /* Storing the SQL for reviewing */ + IO_CACHE review_sql_cache; +#endif } PRINT_EVENT_INFO; #endif @@ -1223,6 +1237,37 @@ public: void print_base64(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info, bool is_more); #endif + + /* The following code used for Flashback */ +#ifdef MYSQL_CLIENT + my_bool is_flashback; + my_bool need_flashback_review; + String output_buf; // Storing the event output +#ifdef WHEN_FLASHBACK_REVIEW_READY + String m_review_dbname; + String m_review_tablename; + + void set_review_dbname(const char *name) + { + if (name) + { + m_review_dbname.free(); + m_review_dbname.append(name); + } + } + void set_review_tablename(const char *name) + { + if (name) + { + m_review_tablename.free(); + m_review_tablename.append(name); + } + } + const char *get_review_dbname() const { return m_review_dbname.ptr(); } + const char *get_review_tablename() const { return m_review_tablename.ptr(); } +#endif +#endif + /* read_log_event() functions read an event from a binlog or relay log; used by SHOW BINLOG EVENTS, the binlog_dump thread on the @@ -4362,12 +4407,14 @@ public: #ifdef MYSQL_CLIENT /* not for direct call, each derived has its own ::print() */ virtual void print(FILE *file, PRINT_EVENT_INFO *print_event_info)= 0; + void change_to_flashback_event(PRINT_EVENT_INFO *print_event_info, uchar *rows_buff, Log_event_type ev_type); void print_verbose(IO_CACHE *file, PRINT_EVENT_INFO *print_event_info); size_t print_verbose_one_row(IO_CACHE *file, table_def *td, PRINT_EVENT_INFO *print_event_info, MY_BITMAP *cols_bitmap, - const uchar *ptr, const uchar *prefix); + const uchar *ptr, const uchar *prefix, + const my_bool no_fill_output= 0); // if no_fill_output=1, then print result is unnecessary #endif #ifdef MYSQL_SERVER @@ -4506,6 +4553,8 @@ protected: uchar *m_rows_cur; /* One-after the end of the data */ uchar *m_rows_end; /* One-after the end of the allocated space */ + size_t m_rows_before_size; /* The length before m_rows_buf */ + flag_set m_flags; /* Flags for row-level events */ Log_event_type m_type; /* Actual event type */ @@ -5040,6 +5089,29 @@ public: }; +static inline bool copy_event_cache_to_string_and_reinit(IO_CACHE *cache, LEX_STRING *to) +{ + String tmp; + + reinit_io_cache(cache, READ_CACHE, 0L, FALSE, FALSE); + if (tmp.append(cache, cache->end_of_file)) + goto err; + reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE); + + /* + Can't change the order, because the String::release() will clear the + length. + */ + to->length= tmp.length(); + to->str= tmp.release(); + + return false; + +err: + perror("Out of memory: can't allocate memory in copy_event_cache_to_string_and_reinit()."); + return true; +} + static inline bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, FILE *file) { diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 8930bf930d1..23ac568d95e 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -393,7 +393,7 @@ bool opt_bin_log_compress; uint opt_bin_log_compress_min_len; my_bool opt_log, debug_assert_if_crashed_table= 0, opt_help= 0; my_bool debug_assert_on_not_freed_memory= 0; -my_bool disable_log_notes; +my_bool disable_log_notes, opt_support_flashback= 0; static my_bool opt_abort; ulonglong log_output_options; my_bool opt_userstat_running; @@ -7413,6 +7413,10 @@ struct my_option my_long_options[]= 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0}, /* We must always support the next option to make scripts like mysqltest easier to do */ + {"flashback", 0, + "Setup the server to use flashback. This enables binary log in row mode and will enable extra logging for DDL's needed by flashback feature", + &opt_support_flashback, &opt_support_flashback, + 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0}, {"gdb", 0, "Set up signals usable for debugging. Deprecated, use --debug-gdb instead.", &opt_debugging, &opt_debugging, @@ -9587,6 +9591,18 @@ static int get_options(int *argc_ptr, char ***argv_ptr) else global_system_variables.option_bits&= ~OPTION_BIG_SELECTS; + if (opt_support_flashback) + { + /* Force binary logging */ + if (!opt_bin_logname) + opt_bin_logname= (char*) ""; // Use default name + opt_bin_log= opt_bin_log_used= 1; + + /* Force format to row */ + binlog_format_used= 1; + global_system_variables.binlog_format= BINLOG_FORMAT_ROW; + } + if (!opt_bootstrap && WSREP_PROVIDER_EXISTS && global_system_variables.binlog_format != BINLOG_FORMAT_ROW) { diff --git a/sql/mysqld.h b/sql/mysqld.h index 96944c012ce..9022b8cef03 100644 --- a/sql/mysqld.h +++ b/sql/mysqld.h @@ -114,6 +114,7 @@ extern uint opt_bin_log_compress_min_len; extern my_bool opt_log, opt_bootstrap; extern my_bool opt_backup_history_log; extern my_bool opt_backup_progress_log; +extern my_bool opt_support_flashback; extern ulonglong log_output_options; extern ulong log_backup_output_options; extern my_bool opt_log_queries_not_using_indexes; diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc index 56913092354..057e18f8f06 100644 --- a/sql/sys_vars.cc +++ b/sql/sys_vars.cc @@ -449,16 +449,17 @@ static bool binlog_format_check(sys_var *self, THD *thd, set_var *var) /* MariaDB Galera does not support STATEMENT or MIXED binlog format currently. */ - if (WSREP(thd) && var->save_result.ulonglong_value != BINLOG_FORMAT_ROW) + if ((WSREP(thd) || opt_support_flashback) && + var->save_result.ulonglong_value != BINLOG_FORMAT_ROW) { // Push a warning to the error log. push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR, - "MariaDB Galera does not support binlog format: %s", + "MariaDB Galera and flashback does not support binlog format: %s", binlog_format_names[var->save_result.ulonglong_value]); if (var->type == OPT_GLOBAL) { - WSREP_ERROR("MariaDB Galera does not support binlog format: %s", + WSREP_ERROR("MariaDB Galera and flashback does not support binlog format: %s", binlog_format_names[var->save_result.ulonglong_value]); return true; } -- cgit v1.2.1