October 2009 - developers - lists.mariadb.org

[Maria-developers] Warnings in MariaDB
by Michael Widenius 07 Dec '09

07 Dec '09

Hi! I have been removing warnings (found by buildbot) from the MariaDB 5.1 tree. (Will be pushed later today). Here is some comments, suggestions of what we still to do to make the buildbot part 'more clean: ------ We get a lot of warnings of the type: ctype-ucs2.c:206: warning: â€˜s_wcâ€™ may be used uninitialized in this function This are false warnings (The compiler is not smart enough to figure this out). The variables are also protected with the LINT_INIT() macro. Suggestion (for Kristian): - If you compile with -Wuninitialized, also use the -DFORCE_INIT_OF_VARS When compiling for valgrind, we should not use -Wuninitialized or -DFORCE_INIT_OF_VARS ------- For Percona: We get the following warnings from xtradb: ../../storage/xtradb/include/dict0dict.ic:729: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement ../../storage/xtradb/include/dict0dict.ic:788: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement ../../storage/xtradb/include/dict0dict.ic:729: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement ../../storage/xtradb/include/dict0dict.ic:788: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement ../../include/../storage/xtradb/include/dict0dict.ic:729: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement ../../include/../storage/xtradb/include/dict0dict.ic:788: warning: suggest a space before â€˜;â€™ or explicit braces around empty body in â€˜whileâ€™ statement Can you fix these ? ---------- I don't know if the following warning comes from us or from the include files: /usr/include/bits/string3.h:82: warning: call to â€˜__warn_memset_zero_lenâ€™ declared with attribute warning: memset used with constant zero length parameter; this could be due to transposed parameters Need suggestions/help to solve this one. Either we need to find the code in MariaDB that generates the warning or disable warnings in buildbot from /usr/include --------- Regards, Monty

2 1

[Maria-developers] bzr commit into Mariadb 5.2, with Maria 2.0:maria/5.2 branch (sanja:2725)
by sanja＠askmonty.org 27 Nov '09

27 Nov '09

#At lp:maria/5.2 2725 sanja(a)askmonty.org 2009-07-07 Group commit and small optimisation of flush log (in case of page without CRC and sector protection) added. (for review) modified: mysql-test/suite/maria/r/maria3.result storage/maria/ha_maria.cc storage/maria/ma_init.c storage/maria/ma_loghandler.c storage/maria/ma_loghandler.h per-file messages: mysql-test/suite/maria/r/maria3.result New variables added storage/maria/ha_maria.cc New variables for group commit and routines to control them added. storage/maria/ma_init.c Service thread stop (if it is present) storage/maria/ma_loghandler.c 2 types of group commit added. If page is not protetcted by CRC or sector protection (i.e. we do not change header when add data to the page) we do not copy beginning of the page during forcing the buffer where the page is last to close. storage/maria/ma_loghandler.h Routines to control group commit added. === modified file 'mysql-test/suite/maria/r/maria3.result' --- a/mysql-test/suite/maria/r/maria3.result 2009-02-19 09:01:25 +0000 +++ b/mysql-test/suite/maria/r/maria3.result 2009-07-07 00:37:23 +0000 @@ -264,6 +264,8 @@ Variable_name Value maria_block_size 8192 maria_checkpoint_interval 30 maria_force_start_after_recovery_failures 0 +maria_group_commit none +maria_group_commit_rate 800 maria_log_file_size 4294959104 maria_log_purge_type immediate maria_max_sort_file_size 9223372036853727232 @@ -285,6 +287,7 @@ Maria_pagecache_read_requests # Maria_pagecache_reads # Maria_pagecache_write_requests # Maria_pagecache_writes # +Maria_transaction_log_syncs # create table t1 (b char(0)); insert into t1 values(NULL),(""); select length(b) from t1; === modified file 'storage/maria/ha_maria.cc' --- a/storage/maria/ha_maria.cc 2009-05-19 09:28:05 +0000 +++ b/storage/maria/ha_maria.cc 2009-07-07 00:37:23 +0000 @@ -101,22 +101,40 @@ TYPELIB maria_translog_purge_type_typeli array_elements(maria_translog_purge_type_names) - 1, "", maria_translog_purge_type_names, NULL }; + +/* transactional log directory sync */ const char *maria_sync_log_dir_names[]= { "NEVER", "NEWFILE", "ALWAYS", NullS }; - TYPELIB maria_sync_log_dir_typelib= { array_elements(maria_sync_log_dir_names) - 1, "", maria_sync_log_dir_names, NULL }; +/* transactional log group commit */ +const char *maria_group_commit_names[]= +{ + "none", "hard", "soft", NullS +}; +TYPELIB maria_group_commit_typelib= +{ + array_elements(maria_group_commit_names) - 1, "", + maria_group_commit_names, NULL +}; + /** Interval between background checkpoints in seconds */ static ulong checkpoint_interval; static void update_checkpoint_interval(MYSQL_THD thd, struct st_mysql_sys_var *var, void *var_ptr, const void *save); +static void update_maria_group_commit(MYSQL_THD thd, + struct st_mysql_sys_var *var, + void *var_ptr, const void *save); +static void update_maria_group_commit_rate(MYSQL_THD thd, + struct st_mysql_sys_var *var, + void *var_ptr, const void *save); /** After that many consecutive recovery failures, remove logs */ static ulong force_start_after_recovery_failures; static void update_log_file_size(MYSQL_THD thd, @@ -163,6 +181,24 @@ static MYSQL_SYSVAR_ULONG(log_file_size, NULL, update_log_file_size, TRANSLOG_FILE_SIZE, TRANSLOG_MIN_FILE_SIZE, 0xffffffffL, TRANSLOG_PAGE_SIZE); +static MYSQL_SYSVAR_ENUM(group_commit, maria_group_commit, + PLUGIN_VAR_RQCMDARG, + "Specifies maria group commit mode. " + "Possible values are \"none\" (no group commit), " + "\"hard\" (with waiting to actual commit), " + "\"soft\" (no wait for commit (DANGEROUS!!!))", + NULL, update_maria_group_commit, + TRANSLOG_GCOMMIT_NONE, &maria_group_commit_typelib); + +static MYSQL_SYSVAR_ULONG(group_commit_rate, maria_group_commit_rate, + PLUGIN_VAR_RQCMDARG, + "Number of commits per 100 seconds. (in other words one commit for" + "every 100/maria_group_commit_rate second). 0 stands for no waiting" + "for other threads to come and do a commit in \"hard\" mode and no" + " sync()/commit at all in \"soft\" mode. Option has only an effect" + "if maria_group_commit is used", + NULL, update_maria_group_commit_rate, 800, 0, UINT_MAX, 1); + static MYSQL_SYSVAR_ENUM(log_purge_type, log_purge_type, PLUGIN_VAR_RQCMDARG, "Specifies how maria transactional log will be purged. " @@ -3254,6 +3290,8 @@ static struct st_mysql_sys_var* system_v MYSQL_SYSVAR(block_size), MYSQL_SYSVAR(checkpoint_interval), MYSQL_SYSVAR(force_start_after_recovery_failures), + MYSQL_SYSVAR(group_commit), + MYSQL_SYSVAR(group_commit_rate), MYSQL_SYSVAR(page_checksum), MYSQL_SYSVAR(log_dir_path), MYSQL_SYSVAR(log_file_size), @@ -3284,6 +3322,97 @@ static void update_checkpoint_interval(M } /** + @brief Updates group commit mode +*/ + +static void update_maria_group_commit(MYSQL_THD thd, + struct st_mysql_sys_var *var, + void *var_ptr, const void *save) +{ + ulong value= (ulong)*((long *)var_ptr); + DBUG_ENTER("update_maria_group_commit"); + DBUG_PRINT("enter", ("old value: %lu new value %lu rate %lu", + value, (ulong)(*(long *)save), + maria_group_commit_rate)); + /* old value */ + switch (value) { + case TRANSLOG_GCOMMIT_NONE: + break; + case TRANSLOG_GCOMMIT_HARD: + translog_hard_group_commit(FALSE); + break; + case TRANSLOG_GCOMMIT_SOFT: + translog_soft_sync(FALSE); + if (maria_group_commit_rate) + translog_soft_sync_end(); + break; + default: + DBUG_ASSERT(0); /* impossible */ + } + value= *(ulong *)var_ptr= (ulong)(*(long *)save); + translog_sync(); + /* new value */ + switch (value) { + case TRANSLOG_GCOMMIT_NONE: + break; + case TRANSLOG_GCOMMIT_HARD: + translog_hard_group_commit(TRUE); + break; + case TRANSLOG_GCOMMIT_SOFT: + translog_soft_sync(TRUE); + /* variable change made under global lock so we can just read it */ + if (maria_group_commit_rate) + translog_soft_sync_start(); + break; + default: + DBUG_ASSERT(0); /* impossible */ + } + DBUG_VOID_RETURN; +} + +/** + @brief Updates group commit rate +*/ + +static void update_maria_group_commit_rate(MYSQL_THD thd, + struct st_mysql_sys_var *var, + void *var_ptr, const void *save) +{ + ulong new_value= (ulong)*((long *)save); + ulong *value_ptr= (ulong*) var_ptr; + DBUG_ENTER("update_maria_group_commit_rate"); + DBUG_PRINT("enter", ("old value: %lu new value %lu group commit %lu", + *value_ptr, new_value, maria_group_commit)); + if (new_value && + ((TRANSLOG_RATE_BASE * 1000000000ULL / new_value + + TRANSLOG_RATE_BASE / 2) / + TRANSLOG_RATE_BASE) == 0) + new_value= 0; /* protection against too small value */ + + /* variable change made under global lock so we can just read it */ + switch (maria_group_commit) { + case TRANSLOG_GCOMMIT_NONE: + *value_ptr= new_value; + translog_set_group_commit_rate(new_value); + break; + case TRANSLOG_GCOMMIT_HARD: + *value_ptr= new_value; + translog_set_group_commit_rate(new_value); + break; + case TRANSLOG_GCOMMIT_SOFT: + if (*value_ptr) + translog_soft_sync_end(); + translog_set_group_commit_rate(new_value); + if ((*value_ptr= new_value)) + translog_soft_sync_start(); + break; + default: + DBUG_ASSERT(0); /* impossible */ + } + DBUG_VOID_RETURN; +} + +/** @brief Updates the transaction log file limit. */ @@ -3305,6 +3434,7 @@ static SHOW_VAR status_variables[]= { {"Maria_pagecache_reads", (char*) &maria_pagecache_var.global_cache_read, SHOW_LONGLONG}, {"Maria_pagecache_write_requests", (char*) &maria_pagecache_var.global_cache_w_requests, SHOW_LONGLONG}, {"Maria_pagecache_writes", (char*) &maria_pagecache_var.global_cache_write, SHOW_LONGLONG}, + {"Maria_transaction_log_syncs", (char*) &translog_syncs, SHOW_LONGLONG}, {NullS, NullS, SHOW_LONG} }; === modified file 'storage/maria/ma_init.c' --- a/storage/maria/ma_init.c 2008-10-09 20:03:54 +0000 +++ b/storage/maria/ma_init.c 2009-07-07 00:37:23 +0000 @@ -82,6 +82,11 @@ void maria_end(void) maria_inited= maria_multi_threaded= FALSE; ft_free_stopwords(); ma_checkpoint_end(); + if (translog_status == TRANSLOG_OK) + { + translog_soft_sync_end(); + translog_sync(); + } if ((trid= trnman_get_max_trid()) > max_trid_in_control_file) { /* === modified file 'storage/maria/ma_loghandler.c' --- a/storage/maria/ma_loghandler.c 2009-05-19 09:28:05 +0000 +++ b/storage/maria/ma_loghandler.c 2009-07-07 00:37:23 +0000 @@ -18,6 +18,7 @@ #include "ma_blockrec.h" /* for some constants and in-write hooks */ #include "ma_key_recover.h" /* For some in-write hooks */ #include "ma_checkpoint.h" +#include "ma_servicethread.h" /* On Windows, neither my_open() nor my_sync() work for directories. @@ -47,6 +48,15 @@ #include <m_ctype.h> #endif +/** @brief protects checkpoint_in_progress */ +static pthread_mutex_t LOCK_soft_sync; +/** @brief for killing the background checkpoint thread */ +static pthread_cond_t COND_soft_sync; +/** @brief control structure for checkpoint background thread */ +static MA_SERVICE_THREAD_CONTROL soft_sync_control= + {THREAD_DEAD, FALSE, &LOCK_soft_sync, &COND_soft_sync}; + + /* transaction log file descriptor */ typedef struct st_translog_file { @@ -124,10 +134,20 @@ struct st_translog_buffer /* Previous buffer offset to detect it flush finish */ TRANSLOG_ADDRESS prev_buffer_offset; /* + If the buffer was forced to close it save value of its horizon + otherwise LSN_IMPOSSIBLE + */ + TRANSLOG_ADDRESS pre_force_close_horizon; + /* How much is written (or will be written when copy_to_buffer_in_progress become 0) to this buffer */ translog_size_t size; + /* + How much data was skipped during moving page from previous buffer + to this one (it is optimisation of forcing buffer to finish + */ + uint skipped_data; /* File handler for this buffer */ TRANSLOG_FILE *file; /* Threads which are waiting for buffer filling/freeing */ @@ -304,6 +324,7 @@ struct st_translog_descriptor */ pthread_mutex_t log_flush_lock; pthread_cond_t log_flush_cond; + pthread_cond_t new_goal_cond; /* Protects changing of headers of finished files (max_lsn) */ pthread_mutex_t file_header_lock; @@ -344,13 +365,39 @@ static struct st_translog_descriptor log ulong log_purge_type= TRANSLOG_PURGE_IMMIDIATE; ulong log_file_size= TRANSLOG_FILE_SIZE; +/* sync() of log files directory mode */ ulong sync_log_dir= TRANSLOG_SYNC_DIR_NEWFILE; +ulong maria_group_commit= TRANSLOG_GCOMMIT_NONE; +ulong maria_group_commit_rate= 0; /* Marker for end of log */ static uchar end_of_log= 0; #define END_OF_LOG &end_of_log +/** + Switch for "soft" sync (no real sync() but periodical sync by service + thread) +*/ +static volatile my_bool soft_sync= FALSE; +/** + Switch for "hard" group commit mode +*/ +static volatile my_bool hard_group_commit= FALSE; +/** + File numbers interval which have to be sync() +*/ +static uint32 soft_sync_min= 0; +static uint32 soft_sync_max= 0; +/** + stores interval in nanoseconds/TRANSLOG_RATE_BASE (to + fit into uint32) +*/ +static uint32 group_commit_wait= 0; enum enum_translog_status translog_status= TRANSLOG_UNINITED; +ulonglong translog_syncs= 0; /* Number of sync()s */ + +/* time of last flush */ +static ulonglong flush_start= 0; /* chunk types */ #define TRANSLOG_CHUNK_LSN 0x00 /* 0 chunk refer as LSN (head or tail */ @@ -980,12 +1027,17 @@ static TRANSLOG_FILE *get_logfile_by_num static TRANSLOG_FILE *get_current_logfile() { TRANSLOG_FILE *file; + DBUG_ENTER("get_current_logfile"); rw_rdlock(&log_descriptor.open_files_lock); + DBUG_PRINT("info", ("max_file: %lu min_file: %lu open_files: %lu", + (ulong) log_descriptor.max_file, + (ulong) log_descriptor.min_file, + (ulong) log_descriptor.open_files.elements)); DBUG_ASSERT(log_descriptor.max_file - log_descriptor.min_file + 1 == log_descriptor.open_files.elements); file= *dynamic_element(&log_descriptor.open_files, 0, TRANSLOG_FILE **); rw_unlock(&log_descriptor.open_files_lock); - return (file); + DBUG_RETURN(file); } uchar NEAR maria_trans_file_magic[]= @@ -1069,6 +1121,7 @@ static my_bool translog_write_file_heade static my_bool translog_max_lsn_to_header(File file, LSN lsn) { uchar lsn_buff[LSN_STORE_SIZE]; + my_bool rc; DBUG_ENTER("translog_max_lsn_to_header"); DBUG_PRINT("enter", ("File descriptor: %ld " "lsn: (%lu,0x%lx)", @@ -1077,11 +1130,13 @@ static my_bool translog_max_lsn_to_heade lsn_store(lsn_buff, lsn); - DBUG_RETURN(my_pwrite(file, lsn_buff, - LSN_STORE_SIZE, - (LOG_HEADER_DATA_SIZE - LSN_STORE_SIZE), - log_write_flags) != 0 || - my_sync(file, MYF(MY_WME)) != 0); + if (!(rc= (my_pwrite(file, lsn_buff, + LSN_STORE_SIZE, + (LOG_HEADER_DATA_SIZE - LSN_STORE_SIZE), + log_write_flags) != 0 || + my_sync(file, MYF(MY_WME)) != 0))) + translog_syncs++; + DBUG_RETURN(rc); } @@ -1423,7 +1478,9 @@ LSN translog_get_file_max_lsn_stored(uin static my_bool translog_buffer_init(struct st_translog_buffer *buffer, int num) { DBUG_ENTER("translog_buffer_init"); - buffer->prev_last_lsn= buffer->last_lsn= LSN_IMPOSSIBLE; + buffer->pre_force_close_horizon= + buffer->prev_last_lsn= buffer->last_lsn= + LSN_IMPOSSIBLE; DBUG_PRINT("info", ("last_lsn and prev_last_lsn set to 0 buffer: 0x%lx", (ulong) buffer)); @@ -1435,6 +1492,7 @@ static my_bool translog_buffer_init(stru memset(buffer->buffer, TRANSLOG_FILLER, TRANSLOG_WRITE_BUFFER); /* Buffer size */ buffer->size= 0; + buffer->skipped_data= 0; /* cond of thread which is waiting for buffer filling */ if (pthread_cond_init(&buffer->waiting_filling_buffer, 0)) DBUG_RETURN(1); @@ -1489,7 +1547,10 @@ static my_bool translog_close_log_file(T TODO: sync only we have changed the log */ if (!file->is_sync) + { rc= my_sync(file->handler.file, MYF(MY_WME)); + translog_syncs++; + } rc|= my_close(file->handler.file, MYF(MY_WME)); my_free(file, MYF(0)); return test(rc); @@ -2044,7 +2105,8 @@ static void translog_start_buffer(struct (ulong) LSN_OFFSET(log_descriptor.horizon), (ulong) LSN_OFFSET(log_descriptor.horizon))); DBUG_ASSERT(buffer_no == buffer->buffer_no); - buffer->prev_last_lsn= buffer->last_lsn= LSN_IMPOSSIBLE; + buffer->pre_force_close_horizon= + buffer->prev_last_lsn= buffer->last_lsn= LSN_IMPOSSIBLE; DBUG_PRINT("info", ("last_lsn and prev_last_lsn set to 0 buffer: 0x%lx", (ulong) buffer)); buffer->offset= log_descriptor.horizon; @@ -2052,6 +2114,7 @@ static void translog_start_buffer(struct buffer->file= get_current_logfile(); buffer->overlay= 0; buffer->size= 0; + buffer->skipped_data= 0; translog_cursor_init(cursor, buffer, buffer_no); DBUG_PRINT("info", ("file: #%ld (%d) init cursor #%u: 0x%lx " "chaser: %d Size: %lu (%lu)", @@ -2523,6 +2586,7 @@ static my_bool translog_buffer_flush(str TRANSLOG_ADDRESS offset= buffer->offset; TRANSLOG_FILE *file= buffer->file; uint8 ver= buffer->ver; + uint skipped_data; DBUG_ENTER("translog_buffer_flush"); DBUG_PRINT("enter", ("Buffer: #%u 0x%lx file: %d offset: (%lu,0x%lx) size: %lu", @@ -2557,6 +2621,8 @@ static my_bool translog_buffer_flush(str disk */ file= buffer->file; + skipped_data= buffer->skipped_data; + DBUG_ASSERT(skipped_data < TRANSLOG_PAGE_SIZE); for (i= 0, pg= LSN_OFFSET(buffer->offset) / TRANSLOG_PAGE_SIZE; i < buffer->size; i+= TRANSLOG_PAGE_SIZE, pg++) @@ -2573,13 +2639,16 @@ static my_bool translog_buffer_flush(str DBUG_ASSERT(i + TRANSLOG_PAGE_SIZE <= buffer->size); if (translog_status != TRANSLOG_OK && translog_status != TRANSLOG_SHUTDOWN) DBUG_RETURN(1); - if (pagecache_inject(log_descriptor.pagecache, + if (pagecache_write_part(log_descriptor.pagecache, &file->handler, pg, 3, buffer->buffer + i, PAGECACHE_PLAIN_PAGE, PAGECACHE_LOCK_LEFT_UNLOCKED, - PAGECACHE_PIN_LEFT_UNPINNED, 0, - LSN_IMPOSSIBLE)) + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_WRITE_DONE, 0, + LSN_IMPOSSIBLE, + skipped_data, + TRANSLOG_PAGE_SIZE - skipped_data)) { DBUG_PRINT("error", ("Can't write page (%lu,0x%lx) to pagecache, error: %d", @@ -2589,10 +2658,12 @@ static my_bool translog_buffer_flush(str translog_stop_writing(); DBUG_RETURN(1); } + skipped_data= 0; } file->is_sync= 0; - if (my_pwrite(file->handler.file, buffer->buffer, - buffer->size, LSN_OFFSET(buffer->offset), + if (my_pwrite(file->handler.file, buffer->buffer + buffer->skipped_data, + buffer->size - buffer->skipped_data, + LSN_OFFSET(buffer->offset) + buffer->skipped_data, log_write_flags)) { DBUG_PRINT("error", ("Can't write buffer (%lu,0x%lx) size %lu " @@ -2985,6 +3056,7 @@ restart: uchar *from, *table= NULL; int is_last_unfinished_page; uint last_protected_sector= 0; + uint skipped_data= curr_buffer->skipped_data; TRANSLOG_FILE file_copy; uint8 ver= curr_buffer->ver; translog_wait_for_writers(curr_buffer); @@ -2997,7 +3069,25 @@ restart: } DBUG_ASSERT(LSN_FILE_NO(addr) == LSN_FILE_NO(curr_buffer->offset)); from= curr_buffer->buffer + (addr - curr_buffer->offset); - memcpy(buffer, from, TRANSLOG_PAGE_SIZE); + if (skipped_data > (addr - curr_buffer->offset)) + { + /* + We read page part of which is not present in buffer, + so we should read absent part from file (page cache actually) + */ + file= get_logfile_by_number(file_no); + DBUG_ASSERT(file != NULL); + buffer= pagecache_read(log_descriptor.pagecache, &file->handler, + LSN_OFFSET(addr) / TRANSLOG_PAGE_SIZE, + 3, buffer, + PAGECACHE_PLAIN_PAGE, + PAGECACHE_LOCK_LEFT_UNLOCKED, + NULL); + } + else + skipped_data= 0; /* Read after skipped in buffer data */ + memcpy(buffer + skipped_data, from + skipped_data, + TRANSLOG_PAGE_SIZE - skipped_data); /* We can use copy then in translog_page_validator() because it do not put it permanently somewhere. @@ -3291,6 +3381,7 @@ static my_bool translog_truncate_log(TRA uint32 next_page_offset, page_rest; uint32 i; File fd; + int rc; TRANSLOG_VALIDATOR_DATA data; char path[FN_REFLEN]; uchar page_buff[TRANSLOG_PAGE_SIZE]; @@ -3316,14 +3407,19 @@ static my_bool translog_truncate_log(TRA TRANSLOG_PAGE_SIZE); page_rest= next_page_offset - LSN_OFFSET(addr); memset(page_buff, TRANSLOG_FILLER, page_rest); - if ((fd= open_logfile_by_number_no_cache(LSN_FILE_NO(addr))) < 0 || - ((my_chsize(fd, next_page_offset, TRANSLOG_FILLER, MYF(MY_WME)) || - (page_rest && my_pwrite(fd, page_buff, page_rest, LSN_OFFSET(addr), - log_write_flags)) || - my_sync(fd, MYF(MY_WME))) | - my_close(fd, MYF(MY_WME))) || - (sync_log_dir >= TRANSLOG_SYNC_DIR_ALWAYS && - sync_dir(log_descriptor.directory_fd, MYF(MY_WME | MY_IGNORE_BADFD)))) + rc= ((fd= open_logfile_by_number_no_cache(LSN_FILE_NO(addr))) < 0 || + ((my_chsize(fd, next_page_offset, TRANSLOG_FILLER, MYF(MY_WME)) || + (page_rest && my_pwrite(fd, page_buff, page_rest, LSN_OFFSET(addr), + log_write_flags)) || + my_sync(fd, MYF(MY_WME))))); + translog_syncs++; + rc|= (fd > 0 && my_close(fd, MYF(MY_WME))); + if (sync_log_dir >= TRANSLOG_SYNC_DIR_ALWAYS) + { + rc|= sync_dir(log_descriptor.directory_fd, MYF(MY_WME | MY_IGNORE_BADFD)); + translog_syncs++; + } + if (rc) DBUG_RETURN(1); /* fix the horizon */ @@ -3511,6 +3607,7 @@ my_bool translog_init_with_table(const c pthread_mutex_init(&log_descriptor.dirty_buffer_mask_lock, MY_MUTEX_INIT_FAST) || pthread_cond_init(&log_descriptor.log_flush_cond, 0) || + pthread_cond_init(&log_descriptor.new_goal_cond, 0) || my_rwlock_init(&log_descriptor.open_files_lock, NULL) || my_init_dynamic_array(&log_descriptor.open_files, @@ -3912,7 +4009,6 @@ my_bool translog_init_with_table(const c log_descriptor.flushed= log_descriptor.horizon; log_descriptor.in_buffers_only= log_descriptor.bc.buffer->offset; log_descriptor.max_lsn= LSN_IMPOSSIBLE; /* set to 0 */ - log_descriptor.previous_flush_horizon= log_descriptor.horizon; /* Now 'flushed' is set to 'horizon' value, but 'horizon' is (potentially) address of the next LSN and we want indicate that all LSNs that are @@ -3995,6 +4091,10 @@ my_bool translog_init_with_table(const c It is beginning of the log => there is no LSNs in the log => There is no harm in leaving it "as-is". */ + log_descriptor.previous_flush_horizon= log_descriptor.horizon; + DBUG_PRINT("info", ("previous_flush_horizon: (%lu,0x%lx)", + LSN_IN_PARTS(log_descriptor. + previous_flush_horizon))); DBUG_RETURN(0); } file_no--; @@ -4070,6 +4170,9 @@ my_bool translog_init_with_table(const c translog_free_record_header(&rec); } } + log_descriptor.previous_flush_horizon= log_descriptor.horizon; + DBUG_PRINT("info", ("previous_flush_horizon: (%lu,0x%lx)", + LSN_IN_PARTS(log_descriptor.previous_flush_horizon))); DBUG_RETURN(0); err: ma_message_no_user(0, "log initialization failed"); @@ -4157,6 +4260,7 @@ void translog_destroy() pthread_mutex_destroy(&log_descriptor.log_flush_lock); pthread_mutex_destroy(&log_descriptor.dirty_buffer_mask_lock); pthread_cond_destroy(&log_descriptor.log_flush_cond); + pthread_cond_destroy(&log_descriptor.new_goal_cond); rwlock_destroy(&log_descriptor.open_files_lock); delete_dynamic(&log_descriptor.open_files); delete_dynamic(&log_descriptor.unfinished_files); @@ -6885,11 +6989,11 @@ int translog_read_record_header_from_buf { translog_size_t res; DBUG_ENTER("translog_read_record_header_from_buffer"); + DBUG_PRINT("info", ("page byte: 0x%x offset: %u", + (uint) page[page_offset], (uint) page_offset)); DBUG_ASSERT(translog_is_LSN_chunk(page[page_offset])); DBUG_ASSERT(translog_status == TRANSLOG_OK || translog_status == TRANSLOG_READONLY); - DBUG_PRINT("info", ("page byte: 0x%x offset: %u", - (uint) page[page_offset], (uint) page_offset)); buff->type= (page[page_offset] & TRANSLOG_REC_TYPE); buff->short_trid= uint2korr(page + page_offset + 1); DBUG_PRINT("info", ("Type %u, Short TrID %u, LSN (%lu,0x%lx)", @@ -7356,27 +7460,27 @@ static void translog_force_current_buffe "Buffer addr: (%lu,0x%lx) " "Page addr: (%lu,0x%lx) " "size: %lu (%lu) Pg: %u left: %u in progress %u", - (uint) log_descriptor.bc.buffer_no, - (ulong) log_descriptor.bc.buffer, - LSN_IN_PARTS(log_descriptor.bc.buffer->offset), + (uint) old_buffer_no, + (ulong) old_buffer, + LSN_IN_PARTS(old_buffer->offset), (ulong) LSN_FILE_NO(log_descriptor.horizon), (ulong) (LSN_OFFSET(log_descriptor.horizon) - log_descriptor.bc.current_page_fill), - (ulong) log_descriptor.bc.buffer->size, + (ulong) old_buffer->size, (ulong) (log_descriptor.bc.ptr -log_descriptor.bc. buffer->buffer), (uint) log_descriptor.bc.current_page_fill, (uint) left, - (uint) log_descriptor.bc.buffer-> + (uint) old_buffer-> copy_to_buffer_in_progress)); translog_lock_assert_owner(); LINT_INIT(current_page_fill); - new_buff_beginning= log_descriptor.bc.buffer->offset; - new_buff_beginning+= log_descriptor.bc.buffer->size; /* increase offset */ + new_buff_beginning= old_buffer->offset; + new_buff_beginning+= old_buffer->size; /* increase offset */ DBUG_ASSERT(log_descriptor.bc.ptr !=NULL); DBUG_ASSERT(LSN_FILE_NO(log_descriptor.horizon) == - LSN_FILE_NO(log_descriptor.bc.buffer->offset)); + LSN_FILE_NO(old_buffer->offset)); translog_check_cursor(&log_descriptor.bc); DBUG_ASSERT(left < TRANSLOG_PAGE_SIZE); if (left) @@ -7387,18 +7491,20 @@ static void translog_force_current_buffe */ DBUG_PRINT("info", ("left: %u", (uint) left)); + old_buffer->pre_force_close_horizon= + old_buffer->offset + old_buffer->size; /* decrease offset */ new_buff_beginning-= log_descriptor.bc.current_page_fill; current_page_fill= log_descriptor.bc.current_page_fill; memset(log_descriptor.bc.ptr, TRANSLOG_FILLER, left); - log_descriptor.bc.buffer->size+= left; + old_buffer->size+= left; DBUG_PRINT("info", ("Finish Page buffer #%u: 0x%lx " "Size: %lu", - (uint) log_descriptor.bc.buffer->buffer_no, - (ulong) log_descriptor.bc.buffer, - (ulong) log_descriptor.bc.buffer->size)); - DBUG_ASSERT(log_descriptor.bc.buffer->buffer_no == + (uint) old_buffer->buffer_no, + (ulong) old_buffer, + (ulong) old_buffer->size)); + DBUG_ASSERT(old_buffer->buffer_no == log_descriptor.bc.buffer_no); } else @@ -7509,11 +7615,21 @@ static void translog_force_current_buffe if (left) { - /* - TODO: do not copy beginning of the page if we have no CRC or sector - checks on - */ - memcpy(new_buffer->buffer, data, current_page_fill); + if (log_descriptor.flags & + (TRANSLOG_PAGE_CRC | TRANSLOG_SECTOR_PROTECTION)) + memcpy(new_buffer->buffer, data, current_page_fill); + else + { + /* + This page header does not change if we add more data to the page so + we can not copy it and will not overwrite later + */ + new_buffer->skipped_data= current_page_fill; +#ifndef DBUG_OFF + memset(new_buffer->buffer, 0xa5, current_page_fill); +#endif + DBUG_ASSERT(new_buffer->skipped_data < TRANSLOG_PAGE_SIZE); + } } old_buffer->next_buffer_offset= new_buffer->offset; translog_buffer_lock(new_buffer); @@ -7561,6 +7677,7 @@ void translog_flush_set_new_goal_and_wai { log_descriptor.next_pass_max_lsn= lsn; log_descriptor.max_lsn_requester= pthread_self(); + pthread_cond_broadcast(&log_descriptor.new_goal_cond); } while (flush_no == log_descriptor.flush_no) { @@ -7572,67 +7689,79 @@ void translog_flush_set_new_goal_and_wai /** - @brief Flush the log up to given LSN (included) + @brief sync() range of files (inclusive) and directory (by request) - @param lsn log record serial number up to which (inclusive) - the log has to be flushed + @param min min internal file number to flush + @param max max internal file number to flush + @param sync_dir need sync directory - @return Operation status + return Operation status @retval 0 OK @retval 1 Error - */ -my_bool translog_flush(TRANSLOG_ADDRESS lsn) +static my_bool translog_sync_files(uint32 min, uint32 max, + my_bool sync_dir) { - LSN sent_to_disk= LSN_IMPOSSIBLE; - TRANSLOG_ADDRESS flush_horizon; - uint fn, i; - dirty_buffer_mask_t dirty_buffer_mask; - uint8 last_buffer_no, start_buffer_no; + uint fn; my_bool rc= 0; - DBUG_ENTER("translog_flush"); - DBUG_PRINT("enter", ("Flush up to LSN: (%lu,0x%lx)", LSN_IN_PARTS(lsn))); - DBUG_ASSERT(translog_status == TRANSLOG_OK || - translog_status == TRANSLOG_READONLY); - LINT_INIT(sent_to_disk); - - pthread_mutex_lock(&log_descriptor.log_flush_lock); - DBUG_PRINT("info", ("Everything is flushed up to (%lu,0x%lx)", - LSN_IN_PARTS(log_descriptor.flushed))); - if (cmp_translog_addr(log_descriptor.flushed, lsn) >= 0) + ulonglong flush_interval; + DBUG_ENTER("translog_sync_files"); + DBUG_PRINT("info", ("min: %lu max: %lu sync dir: %d", + (ulong) min, (ulong) max, (int) sync_dir)); + DBUG_ASSERT(min <= max); + + flush_interval= group_commit_wait; + if (flush_interval) + flush_start= my_micro_time(); + for (fn= min; fn <= max; fn++) { - pthread_mutex_unlock(&log_descriptor.log_flush_lock); - DBUG_RETURN(0); - } - if (log_descriptor.flush_in_progress) - { - translog_flush_set_new_goal_and_wait(lsn); - if (!pthread_equal(log_descriptor.max_lsn_requester, pthread_self())) + TRANSLOG_FILE *file= get_logfile_by_number(fn); + DBUG_ASSERT(file != NULL); + if (!file->is_sync) { - /* fix lsn if it was horizon */ - if (cmp_translog_addr(lsn, log_descriptor.bc.buffer->last_lsn) > 0) - lsn= BUFFER_MAX_LSN(log_descriptor.bc.buffer); - translog_flush_wait_for_end(lsn); - pthread_mutex_unlock(&log_descriptor.log_flush_lock); - DBUG_RETURN(0); + if (my_sync(file->handler.file, MYF(MY_WME))) + { + rc= 1; + translog_stop_writing(); + DBUG_RETURN(rc); + } + translog_syncs++; + file->is_sync= 1; } - log_descriptor.next_pass_max_lsn= LSN_IMPOSSIBLE; } - log_descriptor.flush_in_progress= 1; - flush_horizon= log_descriptor.previous_flush_horizon; - DBUG_PRINT("info", ("flush_in_progress is set")); - pthread_mutex_unlock(&log_descriptor.log_flush_lock); - translog_lock(); - if (log_descriptor.is_everything_flushed) + if (sync_dir) { - DBUG_PRINT("info", ("everything is flushed")); - rc= (translog_status == TRANSLOG_READONLY); - translog_unlock(); - goto out; + if (!(rc= sync_dir(log_descriptor.directory_fd, + MYF(MY_WME | MY_IGNORE_BADFD)))) + translog_syncs++; } + DBUG_RETURN(rc); +} + + +/* + @brief Flushes buffers with LSNs in them less or equal address <lsn> + + @param lsn address up to which all LSNs should be flushed, + can be reset to real last LSN address + @parem sent_to_disk returns 'sent to disk' position + @param flush_horizon returns horizon of the flush + + @note About terminology see comment to translog_flush(). +*/ + +void translog_flush_buffers(TRANSLOG_ADDRESS *lsn, + TRANSLOG_ADDRESS *sent_to_disk, + TRANSLOG_ADDRESS *flush_horizon) +{ + dirty_buffer_mask_t dirty_buffer_mask; + uint i; + uint8 last_buffer_no, start_buffer_no; + DBUG_ENTER("translog_flush_buffers"); + /* We will recheck information when will lock buffers one by one so we can use unprotected read here (this is just for @@ -7656,15 +7785,15 @@ my_bool translog_flush(TRANSLOG_ADDRESS /* if LSN up to which we have to flush bigger then maximum LSN of previous buffer and at least one LSN was saved in the current buffer (last_lsn != - LSN_IMPOSSIBLE) then we better finish the current buffer. + LSN_IMPOSSIBLE) then we have to close the current buffer. */ - if (cmp_translog_addr(lsn, log_descriptor.bc.buffer->prev_last_lsn) > 0 && + if (cmp_translog_addr(*lsn, log_descriptor.bc.buffer->prev_last_lsn) > 0 && log_descriptor.bc.buffer->last_lsn != LSN_IMPOSSIBLE) { struct st_translog_buffer *buffer= log_descriptor.bc.buffer; - lsn= log_descriptor.bc.buffer->last_lsn; /* fix lsn if it was horizon */ + *lsn= log_descriptor.bc.buffer->last_lsn; /* fix lsn if it was horizon */ DBUG_PRINT("info", ("LSN to flush fixed to last lsn: (%lu,0x%lx)", - LSN_IN_PARTS(log_descriptor.bc.buffer->last_lsn))); + LSN_IN_PARTS(log_descriptor.bc.buffer->last_lsn))); last_buffer_no= log_descriptor.bc.buffer_no; log_descriptor.is_everything_flushed= 1; translog_force_current_buffer_to_finish(); @@ -7676,8 +7805,10 @@ my_bool translog_flush(TRANSLOG_ADDRESS TRANSLOG_BUFFERS_NO); translog_unlock(); } - sent_to_disk= translog_get_sent_to_disk(); - if (cmp_translog_addr(lsn, sent_to_disk) > 0) + + /* flush buffers */ + *sent_to_disk= translog_get_sent_to_disk(); + if (cmp_translog_addr(*lsn, *sent_to_disk) > 0) { DBUG_PRINT("info", ("Start buffer #: %u last buffer #: %u", @@ -7697,53 +7828,237 @@ my_bool translog_flush(TRANSLOG_ADDRESS LSN_IN_PARTS(buffer->last_lsn), (buffer->file ? "dirty" : "closed"))); - if (buffer->prev_last_lsn <= lsn && + if (buffer->prev_last_lsn <= *lsn && buffer->file != NULL) { - DBUG_ASSERT(flush_horizon <= buffer->offset + buffer->size); - flush_horizon= buffer->offset + buffer->size; + DBUG_ASSERT(*flush_horizon <= buffer->offset + buffer->size); + *flush_horizon= (buffer->pre_force_close_horizon != LSN_IMPOSSIBLE ? + buffer->pre_force_close_horizon : + buffer->offset + buffer->size); + /* pre_force_close_horizon is reset during new buffer start */ + DBUG_PRINT("info", ("flush_horizon: (%lu,0x%lx)", + LSN_IN_PARTS(*flush_horizon))); + DBUG_ASSERT(*flush_horizon <= log_descriptor.horizon); + translog_buffer_flush(buffer); } translog_buffer_unlock(buffer); i= (i + 1) % TRANSLOG_BUFFERS_NO; } while (i != last_buffer_no); - sent_to_disk= translog_get_sent_to_disk(); + *sent_to_disk= translog_get_sent_to_disk(); + } + + DBUG_VOID_RETURN; +} + +/** + @brief Flush the log up to given LSN (included) + + @param lsn log record serial number up to which (inclusive) + the log has to be flushed + + @return Operation status + @retval 0 OK + @retval 1 Error + + @note + + - Non group commit logic: Commits made in passes. Thread which started + flush first is performing actual flush, other threads sets new goal (LSN) + of the next pass (if it is maximum) and waits for the pass end or just + wait for the pass end. + + - If hard group commit enabled and rate set to zero: + The first thread sends all changed buffers to disk. This is repeated + as long as there are new LSNs added. The process can not loop + forever because we have limited number of threads and they will wait + for the data to be synced. + Pseudo code: + + do + send changed buffers to disk + while new_goal + sync + + - If hard group commit switched ON and less than rate microseconds has + passed from last sync, then after buffers have been sent to disk + wait until rate microseconds has passed since last sync, do sync and return. + This ensures that if we call sync infrequently we don't do any waits. + + - If soft group commit enabled everything works as with 'non group commit' + but the thread doesn't do any real sync(). If rate is not zero the + sync() will be performed by a service thread with the given rate + when needed (new LSN appears). + + @note Terminology: + 'sent to disk' means written to disk but not sync()ed, + 'flushed' mean sent to disk and synced(). +*/ + +my_bool translog_flush(TRANSLOG_ADDRESS lsn) +{ + struct timespec abstime; + ulonglong flush_interval; + ulonglong time_spent; + LSN sent_to_disk= LSN_IMPOSSIBLE; + TRANSLOG_ADDRESS flush_horizon; + my_bool rc= 0; + my_bool hgroup_commit_at_start; + DBUG_ENTER("translog_flush"); + DBUG_PRINT("enter", ("Flush up to LSN: (%lu,0x%lx)", LSN_IN_PARTS(lsn))); + DBUG_ASSERT(translog_status == TRANSLOG_OK || + translog_status == TRANSLOG_READONLY); + LINT_INIT(sent_to_disk); + + pthread_mutex_lock(&log_descriptor.log_flush_lock); + DBUG_PRINT("info", ("Everything is flushed up to (%lu,0x%lx)", + LSN_IN_PARTS(log_descriptor.flushed))); + if (cmp_translog_addr(log_descriptor.flushed, lsn) >= 0) + + + { + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + DBUG_RETURN(0); } + if (log_descriptor.flush_in_progress) + { + translog_lock(); + /* fix lsn if it was horizon */ + if (cmp_translog_addr(lsn, log_descriptor.bc.buffer->last_lsn) > 0) + lsn= BUFFER_MAX_LSN(log_descriptor.bc.buffer); + translog_unlock(); + translog_flush_set_new_goal_and_wait(lsn); + if (!pthread_equal(log_descriptor.max_lsn_requester, pthread_self())) + { + /* + translog_flush_wait_for_end() release log_flush_lock while is + waiting then acquire it again + */ + translog_flush_wait_for_end(lsn); + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + DBUG_RETURN(0); + } + log_descriptor.next_pass_max_lsn= LSN_IMPOSSIBLE; + } + log_descriptor.flush_in_progress= 1; + flush_horizon= log_descriptor.previous_flush_horizon; + DBUG_PRINT("info", ("flush_in_progress is set, flush_horizon: (%lu,0x%lx)", + LSN_IN_PARTS(flush_horizon))); + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + + hgroup_commit_at_start= hard_group_commit; + if (hgroup_commit_at_start) + flush_interval= group_commit_wait * TRANSLOG_RATE_BASE; - /* sync files from previous flush till current one */ - for (fn= LSN_FILE_NO(log_descriptor.flushed); fn <= LSN_FILE_NO(lsn); fn++) + translog_lock(); + if (log_descriptor.is_everything_flushed) { - TRANSLOG_FILE *file= get_logfile_by_number(fn); - DBUG_ASSERT(file != NULL); - if (!file->is_sync) + DBUG_PRINT("info", ("everything is flushed")); + translog_unlock(); + pthread_mutex_lock(&log_descriptor.log_flush_lock); + goto out; + } + + for (;;) + { + /* Following function flushes buffers and makes translog_unlock() */ + translog_flush_buffers(&lsn, &sent_to_disk, &flush_horizon); + + if (!hgroup_commit_at_start) + break; /* flush pass is ended */ + +retest: + if (flush_interval != 0 && + (my_micro_time() - flush_start) >= flush_interval) + break; /* flush pass is ended */ + + pthread_mutex_lock(&log_descriptor.log_flush_lock); + if (log_descriptor.next_pass_max_lsn != LSN_IMPOSSIBLE) + { + /* take next goal */ + lsn= log_descriptor.next_pass_max_lsn; + log_descriptor.next_pass_max_lsn= LSN_IMPOSSIBLE; + /* prevent other thread from continue */ + log_descriptor.max_lsn_requester= pthread_self(); + DBUG_PRINT("info", ("flush took next goal: (%lu,0x%lx)", + LSN_IN_PARTS(lsn))); + } + else { - if (my_sync(file->handler.file, MYF(MY_WME))) + if (flush_interval == 0 || + (time_spent= (my_micro_time() - flush_start)) >= flush_interval) { - rc= 1; - translog_stop_writing(); - sent_to_disk= LSN_IMPOSSIBLE; - goto out; + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + break; } - file->is_sync= 1; + DBUG_PRINT("info", ("flush waits: %llu interval: %llu spent: %llu", + flush_interval - time_spent, + flush_interval, time_spent)); + /* wait time or next goal */ + set_timespec_nsec(abstime, flush_interval - time_spent); + pthread_cond_timedwait(&log_descriptor.new_goal_cond, + &log_descriptor.log_flush_lock, + &abstime); + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + DBUG_PRINT("info", ("retest conditions")); + goto retest; } + pthread_mutex_unlock(&log_descriptor.log_flush_lock); + + /* next flush pass */ + DBUG_PRINT("info", ("next flush pass")); + translog_lock(); } - if (sync_log_dir >= TRANSLOG_SYNC_DIR_ALWAYS && - (LSN_FILE_NO(log_descriptor.previous_flush_horizon) != - LSN_FILE_NO(flush_horizon) || - ((LSN_OFFSET(log_descriptor.previous_flush_horizon) - 1) / - TRANSLOG_PAGE_SIZE) != - ((LSN_OFFSET(flush_horizon) - 1) / TRANSLOG_PAGE_SIZE))) - rc|= sync_dir(log_descriptor.directory_fd, MYF(MY_WME | MY_IGNORE_BADFD)); + /* + sync() files from previous flush till current one + */ + if (!soft_sync || hgroup_commit_at_start) + { + if ((rc= + translog_sync_files(LSN_FILE_NO(log_descriptor.flushed), + LSN_FILE_NO(lsn), + sync_log_dir >= TRANSLOG_SYNC_DIR_ALWAYS && + (LSN_FILE_NO(log_descriptor. + previous_flush_horizon) != + LSN_FILE_NO(flush_horizon) || + (LSN_OFFSET(log_descriptor. + previous_flush_horizon) / + TRANSLOG_PAGE_SIZE) != + (LSN_OFFSET(flush_horizon) / + TRANSLOG_PAGE_SIZE))))) + { + sent_to_disk= LSN_IMPOSSIBLE; + pthread_mutex_lock(&log_descriptor.log_flush_lock); + goto out; + } + /* keep values for soft sync() and forced sync() actual */ + { + uint32 fileno= LSN_FILE_NO(lsn); + my_atomic_rwlock_wrlock(&soft_sync_rwl); + my_atomic_store32(&soft_sync_min, fileno); + my_atomic_store32(&soft_sync_max, fileno); + my_atomic_rwlock_wrunlock(&soft_sync_rwl); + } + } + else + { + my_atomic_rwlock_wrlock(&soft_sync_rwl); + my_atomic_store32(&soft_sync_max, LSN_FILE_NO(lsn)); + my_atomic_rwlock_wrunlock(&soft_sync_rwl); + } + + DBUG_ASSERT(flush_horizon <= log_descriptor.horizon); + + pthread_mutex_lock(&log_descriptor.log_flush_lock); log_descriptor.previous_flush_horizon= flush_horizon; out: - pthread_mutex_lock(&log_descriptor.log_flush_lock); if (sent_to_disk != LSN_IMPOSSIBLE) log_descriptor.flushed= sent_to_disk; log_descriptor.flush_in_progress= 0; log_descriptor.flush_no++; DBUG_PRINT("info", ("flush_in_progress is dropped")); - pthread_mutex_unlock(&log_descriptor.log_flush_lock);\ + pthread_mutex_unlock(&log_descriptor.log_flush_lock); pthread_cond_broadcast(&log_descriptor.log_flush_cond); DBUG_RETURN(rc); } @@ -8113,6 +8428,8 @@ LSN translog_first_theoretical_lsn() my_bool translog_purge(TRANSLOG_ADDRESS low) { uint32 last_need_file= LSN_FILE_NO(low); + uint32 min_unsync; + int soft; TRANSLOG_ADDRESS horizon= translog_get_horizon(); int rc= 0; DBUG_ENTER("translog_purge"); @@ -8120,12 +8437,23 @@ my_bool translog_purge(TRANSLOG_ADDRESS DBUG_ASSERT(translog_status == TRANSLOG_OK || translog_status == TRANSLOG_READONLY); + soft= soft_sync; + DBUG_PRINT("info", ("min_unsync: %lu", (ulong) min_unsync)); + if (soft && min_unsync < last_need_file) + { + last_need_file= min_unsync; + DBUG_PRINT("info", ("last_need_file set to %lu", (ulong)last_need_file)); + } + pthread_mutex_lock(&log_descriptor.purger_lock); + DBUG_PRINT("info", ("last_lsn_checked file: %lu:", + (ulong) log_descriptor.last_lsn_checked)); if (LSN_FILE_NO(log_descriptor.last_lsn_checked) < last_need_file) { uint32 i; uint32 min_file= translog_first_file(horizon, 1); DBUG_ASSERT(min_file != 0); /* log is already started */ + DBUG_PRINT("info", ("min_file: %lu:",(ulong) min_file)); for(i= min_file; i < last_need_file && rc == 0; i++) { LSN lsn= translog_get_file_max_lsn_stored(i); @@ -8356,6 +8684,153 @@ my_bool translog_log_debug_info(TRN *trn } + +/** + Sets soft sync mode + + @param mode TRUE if we need switch soft sync on else off +*/ + +void translog_soft_sync(my_bool mode) +{ + soft_sync= mode; +} + + +/** + Sets hard group commit + + @param mode TRUE if we need switch hard group commit on else off +*/ + +void translog_hard_group_commit(my_bool mode) +{ + hard_group_commit= mode; +} + + +/** + @brief forced log sync (used when we are switching modes) +*/ + +void translog_sync() +{ + uint32 max= get_current_logfile()->number; + uint32 min; + DBUG_ENTER("ma_translog_sync"); + + my_atomic_rwlock_rdlock(&soft_sync_rwl); + min= my_atomic_load32(&soft_sync_min); + my_atomic_rwlock_rdunlock(&soft_sync_rwl); + if (!min) + min= max; + + translog_sync_files(min, max, sync_log_dir >= TRANSLOG_SYNC_DIR_ALWAYS); + + DBUG_VOID_RETURN; +} + + +/** + @brief set rate for group commit + + @param rate rate to set. + + @note internally it stores interval in nanoseconds/TRANSLOG_RATE_BASE (to + fit into uint32) +*/ + +void translog_set_group_commit_rate(uint32 rate) +{ + DBUG_ENTER("translog_set_group_commit_rate"); + ulonglong wait_time; + if (rate) + { + wait_time= ((TRANSLOG_RATE_BASE * 1000000000ULL / rate + + TRANSLOG_RATE_BASE / 2) / + TRANSLOG_RATE_BASE); + if (wait_time == 0) + wait_time= 1; /* protection from getting special value */ + } + else + wait_time= 0; + group_commit_wait= wait_time; + DBUG_PRINT("info", ("rate: %lu wait: %llu", + (ulong)rate, (ulonglong)wait_time)); + DBUG_VOID_RETURN; +} + + +/** + @brief syncing service thread +*/ + +static pthread_handler_t +ma_soft_sync_background( void *arg __attribute__((unused))) +{ + + my_thread_init(); + { + DBUG_ENTER("ma_soft_sync_background"); + for(;;) + { + ulonglong prev_loop= my_micro_time(); + ulonglong time, sleep; + uint32 min, max; + my_atomic_rwlock_rdlock(&soft_sync_rwl); + min= my_atomic_load32(&soft_sync_min); + max= my_atomic_load32(&soft_sync_max); + my_atomic_store32(&soft_sync_min, max); + my_atomic_rwlock_rdunlock(&soft_sync_rwl); + + sleep= group_commit_wait * TRANSLOG_RATE_BASE; + translog_sync_files(min, max, FALSE); + time= my_micro_time() - prev_loop; + if (time > sleep) + sleep= 0; + else + sleep-= time; + if (my_service_thread_sleep(&soft_sync_control, sleep)) + break; + } + my_service_thread_signal_end(&soft_sync_control); + my_thread_end(); + DBUG_RETURN(0); + } +} + + +/** + @brief Starts syncing thread +*/ + +int translog_soft_sync_start(void) +{ + pthread_t th; + int res= 0; + DBUG_ENTER("translog_soft_sync_start"); + if (!(res= ma_service_thread_control_init(&soft_sync_control))) + if (!(res= pthread_create(&th, NULL, ma_soft_sync_background, NULL))) + soft_sync_control.status= THREAD_RUNNING; + DBUG_RETURN(res); +} + + +/** + @brief Stops syncing thread +*/ + +void translog_soft_sync_end(void) +{ + DBUG_ENTER("translog_soft_sync_end"); + if (soft_sync_control.inited) + { + ma_service_thread_control_end(&soft_sync_control); + } + DBUG_VOID_RETURN; +} + + #ifdef MARIA_DUMP_LOG #include <my_getopt.h> extern void translog_example_table_init(); === modified file 'storage/maria/ma_loghandler.h' --- a/storage/maria/ma_loghandler.h 2009-01-15 22:25:53 +0000 +++ b/storage/maria/ma_loghandler.h 2009-07-07 00:37:23 +0000 @@ -342,6 +342,14 @@ enum enum_translog_status TRANSLOG_SHUTDOWN /* going to shutdown the loghandler */ }; extern enum enum_translog_status translog_status; +extern ulonglong translog_syncs; /* Number of sync()s */ + +void translog_soft_sync(my_bool mode); +void translog_hard_group_commit(my_bool mode); +int translog_soft_sync_start(void); +void translog_soft_sync_end(void); +void translog_sync(); +void translog_set_group_commit_rate(uint32 rate); /* all the rest added because of recovery; should we make @@ -441,6 +449,18 @@ extern LOG_DESC log_record_type_descript typedef enum { + TRANSLOG_GCOMMIT_NONE, + TRANSLOG_GCOMMIT_HARD, + TRANSLOG_GCOMMIT_SOFT +} enum_maria_group_commit; +extern ulong maria_group_commit; +extern ulong maria_group_commit_rate; +/** + group commit interval is TRANSLOG_RATE_BASE/<rate> seconds +*/ +#define TRANSLOG_RATE_BASE 100 +typedef enum +{ TRANSLOG_PURGE_IMMIDIATE, TRANSLOG_PURGE_EXTERNAL, TRANSLOG_PURGE_ONDEMAND

2 1

[Maria-developers] request for comment: OQGRAPH in 5.1 packages
by Arjen Lentz 09 Nov '09

09 Nov '09

Hi all, fellow Maria captains in particular (but naturally anybody here can comment) We'd like to include the plugin for OQGRAPH engine in the 5.1 packages we're just about to build. It would not be pulled in like the xtradb/pbxt engines, but be compiled separately and not loaded in by default (people will have to do INSTALL PLUGIN) so that as long as it's not loaded, it can have no influence on the running of mysqld. The rationale is this. From the experience with PBXT, people really want/need binaries/ packages before they will try things. If 5.1 binaries had had PBXT plugin sitting there, lots more people would have tried it earlier, filed bugreports and feedback, and Paul would have been where he is now much quicker. With MariaDB pulling it in it's ok now, but it's just a darn waste and pity of the earlier time. Since 5.1's plugin infrastructure still requires a plugin to be compiled against close to exact the original mysqld source, the only way to ensure that is to compile them from the same source at the same time, next to eachother. So that's what I'm proposing. Nonsense like the feature preview builds that Sun/MySQL did just make no sense in the real world, people can't use that. So while sticking new plugins in a future version like 5.2 appears sensible, it doesn't actually help in getting the code out there and used which is of course the only way to get feedback and bugreports. The ability to have plugins distributed but not loaded is the key here, it allows us to get stuff out and those who want to try it can, without destabilising anything for those who don't. (On the practical side, since it's essentially separate it would get added during the source tarball prep in the builds, so no action required inside the maria tree) With the packages getting prepped now, I realise it's a tad short notice. Don't feel rushed, but do please reply when you read this, otherwise it might get lost in your email pile (I know my friends, as I know myself ;-) Thanks Cheers, Arjen.

7 11

[Maria-developers] Rewrite of the free documentation to make it about MariaDB, not MySQL
by Henrik Ingo 03 Nov '09

03 Nov '09

Daniel I glanced through some of the commits today and noticed Kristian committing the new version of MySQL 5.1.38 documentation (man pages and other free docs). The docs seem to include lots of references to where you can download MySQL rpm's, where you can buy MySQL support, Suns support lifecycle policies, etc. As part of your documentation work, could you be in touch with Kristian and find out how you could help in cleaning that up. Meaning: Either remove such sections, or replace with relevant MariaDB information. While at it, I'd like to review the Readme file of MariaDB. Could you help me with that (ie extract it from bzr and work with me to rewrite it - so that I can just participate by email). The Readme file we should fix before GA release. The other docs are only semi-urgent. henrik -- email: henrik.ingo(a)avoinelama.fi tel: +358-40-5697354 www: www.avoinelama.fi/~hingo book: www.openlife.cc

3 5

[Maria-developers] FYI: maria-5.1-merge renamed to 5.1-release
by Sergey Petrunya 02 Nov '09

02 Nov '09

Hi! This is to inform that MariaDB release branch tree, which was at a location with a counter-intuitive name of lp:~maria-captains/maria/maria-5.1-merge has been moved to lp:~maria-captains/maria/5.1-release Buildbot has been repointed to the new tree. BR Sergey -- Sergey Petrunia, Software Developer Monty Program AB, http://askmonty.org Blog: http://s.petrunia.net/blog

2 1

[Maria-developers] MWL#36: feedback for post-review fixes
by Sergey Petrunya 31 Oct '09

31 Oct '09

Hi Alexander, Please find the feedback for post-review fixes below: > === modified file 'client/mysqlbinlog.cc' > --- client/mysqlbinlog.cc 2009-10-24 19:43:39 +0000 > +++ client/mysqlbinlog.cc 2009-10-27 13:42:47 +0000 > @@ -38,8 +38,8 @@ > /* Needed for Rpl_filter */ > CHARSET_INFO* system_charset_info= &my_charset_utf8_general_ci; > > -#include "sql_string.h" // needed for Rpl_filter > -#include "sql_list.h" // needed for Rpl_filter > +#include "sql_string.h" // needed for Rpl_filter > +#include "sql_list.h" // needed for Rpl_filter > #include "rpl_filter.h" > > Rpl_filter *binlog_filter; > @@ -1190,7 +1190,8 @@ > (uchar**) &open_files_limit, (uchar**) &open_files_limit, 0, GET_ULONG, > REQUIRED_ARG, MY_NFILE, 8, OS_FILE_LIMIT, 0, 1, 0}, > {"rewrite-db", OPT_REWRITE_DB, > - "Updates to a database with a different name than the original.", > + "Updates to a database with a different name than the original. \ > +Example: rewrite-db='from->to'.", Please change to use multiple-strings; "Updates to a database with a different name than the original." "Example: rewrite-db='from->to'.", as that is what is done around the codebase. > 0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0}, > {0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0} > }; > > === modified file 'mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_options.result' > --- mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_options.result 2009-10-24 21:48:58 +0000 > +++ mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_options.result 2009-10-27 13:42:47 +0000 > @@ -216,6 +216,201 @@ > # End of log file > ROLLBACK /* added by mysqlbinlog */; > /*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/; > +# > +# mysqlbinlog output > +# --base64-output = decode-rows > +# --rewrite-db = test1->new_test1 > +# --rewrite-db = test3->new_test3 > +# --read-from-remote-server > +# > +/*!40019 SET @@session.max_insert_delayed_threads=0*/; > +/*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/; > +DELIMITER /*!*/; > +# at # > +#010909 4:46:40 server id # end_log_pos # Start: binlog v 4, server v #.##.## created 010909 4:46:40 at startup Apparently this testcase will succeed only at 4:40 on 01-Sept-2009. Please make a stable testcase. > +ROLLBACK/*!*/; > +# at # > +use new_test1/*!*/; > +#010909 4:46:40 server id # end_log_pos # Query thread_id=# exec_time=# error_code=0 > +SET TIMESTAMP=1000000000/*!*/; > +SET @@session.pseudo_thread_id=#/*!*/; After the above is fixed, we'll need to do another push and buildbot run to see if there are any problems. BR Sergey -- Sergey Petrunia, Software Developer Monty Program AB, http://askmonty.org Blog: http://s.petrunia.net/blog

3 4

[Maria-developers] [Branch ~maria-captains/maria/5.1] Rev 2754: Merge with 5.1-release
by noreply＠launchpad.net 30 Oct '09

30 Oct '09

Merge authors: Michael Widenius (monty) Sergey Petrunia (sergefp) ------------------------------------------------------------ revno: 2754 [merge] committer: Michael Widenius <monty(a)askmonty.org> branch nick: maria-5.1 timestamp: Fri 2009-10-30 22:28:11 +0200 message: Merge with 5.1-release modified: scripts/make_binary_distribution.sh win/make_mariadb_win_dist -- lp:maria https://code.launchpad.net/~maria-captains/maria/5.1 Your team Maria developers is subscribed to branch lp:maria. To unsubscribe from this branch go to https://code.launchpad.net/~maria-captains/maria/5.1/+edit-subscription.

1 0

[Maria-developers] bzr commit into MariaDB 5.1, with Maria 1.5:maria branch (monty:2754)
by Michael Widenius 30 Oct '09

30 Oct '09

#At lp:maria based on revid:monty@askmonty.org-20091030201608-t2p53ssllnmcvufz 2754 Michael Widenius 2009-10-30 [merge] Merge with 5.1-release modified: scripts/make_binary_distribution.sh win/make_mariadb_win_dist === modified file 'scripts/make_binary_distribution.sh' --- a/scripts/make_binary_distribution.sh 2009-10-29 00:04:56 +0000 +++ b/scripts/make_binary_distribution.sh 2009-10-30 20:28:11 +0000 @@ -305,11 +305,6 @@ if [ x"$BASE_SYSTEM" != x"netware" ] ; t fi fi - # FIXME let this script be in "bin/", where it is in the RPMs? - # http://dev.mysql.com/doc/refman/5.1/en/mysql-install-db-problems.html - mkdir $DEST/scripts - mv $DEST/bin/mysql_install_db $DEST/scripts/ - # Note, no legacy "safe_mysqld" link to "mysqld_safe" in 5.1 # Copy readme and license files @@ -342,18 +337,25 @@ if [ x"$BASE_SYSTEM" != x"netware" ] ; t # # Move things to make them easier to find in tar installation # - mv $DEST/libexec/* $DEST/bin + + # The following test is needed if the original configure was done with + # something like --libexecdir=/usr/local/mysql/bin + if test -f $DEST/libexec/mysqld + then + mv $DEST/libexec/* $DEST/bin + rmdir $DEST/libexec + fi mv $DEST/share/man $DEST mv $DEST/share/mysql/binary-configure $DEST/configure mv $DEST/share/mysql/*.sql $DEST/share mv $DEST/share/mysql/*.cnf $DEST/share/mysql/*.server $DEST/share/mysql/mysql-log-rotate $DEST/support-files - rmdir $DEST/libexec # # Move some scripts that are only run once to 'scripts' directory # but add symbolic links instead to old place for compatibility # - for i in mysql_secure_installation mysql_fix_extensions mysql_fix_privilege_tables + mkdir $DEST/scripts + for i in mysql_secure_installation mysql_fix_extensions mysql_fix_privilege_tables mysql_install_db do mv $DEST/bin/$i $DEST/scripts ln -s "../scripts/$i" $DEST/bin/$i === modified file 'win/make_mariadb_win_dist' --- a/win/make_mariadb_win_dist 2009-10-12 16:50:20 +0000 +++ b/win/make_mariadb_win_dist 2009-10-30 10:50:48 +0000 @@ -66,7 +66,7 @@ ls -lah $ZIPFILE echo "$ZIPFILE is the Windows noinstall binary zip" if [ $RES ] ; then - echo "Archive contents differ from the standard file list, check the diff output above" + echo "Archive contents differ from the standard file list, check the diff output above" else echo "Archive contents match the standard list, OK" fi

1 0

[Maria-developers] [Branch ~maria-captains/maria/5.1] Rev 2753: Merge of FederatedX code
by noreply＠launchpad.net 30 Oct '09

30 Oct '09

Merge authors: Michael Widenius (monty) ------------------------------------------------------------ revno: 2753 [merge] committer: Michael Widenius <monty(a)askmonty.org> branch nick: maria-5.1 timestamp: Fri 2009-10-30 22:16:08 +0200 message: Merge of FederatedX code added: storage/federated/README storage/federatedx/ storage/federatedx/AUTHORS storage/federatedx/CMakeFiles.txt storage/federatedx/ChangeLog storage/federatedx/FAQ storage/federatedx/Makefile.am storage/federatedx/README storage/federatedx/README.windows storage/federatedx/TODO storage/federatedx/federatedx_io.cc storage/federatedx/federatedx_io_mysql.cc storage/federatedx/federatedx_io_null.cc storage/federatedx/federatedx_probes.h storage/federatedx/federatedx_txn.cc storage/federatedx/ha_federatedx.cc storage/federatedx/ha_federatedx.h storage/federatedx/plug.in renamed: storage/federated/plug.in => storage/federated/plug.in.disabled modified: .bzrignore client/mysqladmin.cc extra/yassl/taocrypt/src/twofish.cpp libmysqld/Makefile.am mysql-test/mysql-test-run.pl mysql-test/suite/federated/disabled.def mysql-test/suite/federated/federated.result mysql-test/suite/federated/federated.test mysql-test/suite/federated/federated_archive.result mysql-test/suite/federated/federated_bug_13118.result mysql-test/suite/federated/federated_bug_25714.result mysql-test/suite/federated/federated_cleanup.inc mysql-test/suite/federated/federated_innodb.result mysql-test/suite/federated/federated_server.result mysql-test/suite/federated/federated_server.test mysql-test/suite/federated/federated_transactions.result mysql-test/valgrind.supp storage/pbxt/src/cache_xt.cc storage/xtradb/include/buf0buf.ic The size of the diff (6093 lines) is larger than your specified limit of 5000 lines -- lp:maria https://code.launchpad.net/~maria-captains/maria/5.1 Your team Maria developers is subscribed to branch lp:maria. To unsubscribe from this branch go to https://code.launchpad.net/~maria-captains/maria/5.1/+edit-subscription.

1 0

[Maria-developers] bzr commit into MariaDB 5.1, with Maria 1.5:maria branch (monty:2753)
by Michael Widenius 30 Oct '09

30 Oct '09

#At lp:maria based on revid:psergey@askmonty.org-20091029175033-4agewzfel760eq7v 2753 Michael Widenius 2009-10-30 [merge] Merge of FederatedX code added: storage/federated/README storage/federatedx/ storage/federatedx/AUTHORS storage/federatedx/CMakeFiles.txt storage/federatedx/ChangeLog storage/federatedx/FAQ storage/federatedx/Makefile.am storage/federatedx/README storage/federatedx/README.windows storage/federatedx/TODO storage/federatedx/federatedx_io.cc storage/federatedx/federatedx_io_mysql.cc storage/federatedx/federatedx_io_null.cc storage/federatedx/federatedx_probes.h storage/federatedx/federatedx_txn.cc storage/federatedx/ha_federatedx.cc storage/federatedx/ha_federatedx.h storage/federatedx/plug.in renamed: storage/federated/plug.in => storage/federated/plug.in.disabled modified: .bzrignore client/mysqladmin.cc extra/yassl/taocrypt/src/twofish.cpp libmysqld/Makefile.am mysql-test/mysql-test-run.pl mysql-test/suite/federated/disabled.def mysql-test/suite/federated/federated.result mysql-test/suite/federated/federated.test mysql-test/suite/federated/federated_archive.result mysql-test/suite/federated/federated_bug_13118.result mysql-test/suite/federated/federated_bug_25714.result mysql-test/suite/federated/federated_cleanup.inc mysql-test/suite/federated/federated_innodb.result mysql-test/suite/federated/federated_server.result mysql-test/suite/federated/federated_server.test mysql-test/suite/federated/federated_transactions.result mysql-test/valgrind.supp storage/pbxt/src/cache_xt.cc storage/xtradb/include/buf0buf.ic === modified file '.bzrignore' --- a/.bzrignore 2009-09-15 12:12:51 +0000 +++ b/.bzrignore 2009-10-30 18:51:46 +0000 @@ -1921,3 +1921,4 @@ sql/share/ukrainian libmysqld/examples/mysqltest.cc extra/libevent/event-config.h libmysqld/opt_table_elimination.cc +libmysqld/ha_federatedx.cc === modified file 'client/mysqladmin.cc' --- a/client/mysqladmin.cc 2009-10-26 11:35:42 +0000 +++ b/client/mysqladmin.cc 2009-10-30 18:50:56 +0000 @@ -1043,7 +1043,7 @@ static int drop_db(MYSQL *mysql, const c printf("Do you really want to drop the '%s' database [y/N] ",db); fflush(stdout); if (fgets(buf,sizeof(buf)-1,stdin) == 0 || - (*buf != 'y') && (*buf != 'Y')) + ((*buf != 'y') && (*buf != 'Y'))) { puts("\nOK, aborting database drop!"); return -1; === modified file 'extra/yassl/taocrypt/src/twofish.cpp' --- a/extra/yassl/taocrypt/src/twofish.cpp 2007-01-29 15:54:40 +0000 +++ b/extra/yassl/taocrypt/src/twofish.cpp 2009-10-30 18:50:56 +0000 @@ -55,6 +55,7 @@ void Twofish::Process(byte* out, const b in += BLOCK_SIZE; } else if (mode_ == CBC) + { if (dir_ == ENCRYPTION) while (blocks--) { r_[0] ^= *(word32*)in; @@ -82,6 +83,7 @@ void Twofish::Process(byte* out, const b out += BLOCK_SIZE; in += BLOCK_SIZE; } + } } #endif // DO_TWOFISH_ASM === modified file 'libmysqld/Makefile.am' --- a/libmysqld/Makefile.am 2009-09-15 10:46:35 +0000 +++ b/libmysqld/Makefile.am 2009-10-30 18:50:56 +0000 @@ -124,7 +124,7 @@ handler.o: handler.cc # found to append fileslists that collected by configure # to the sources list -ha_federated.o:ha_federated.cc +ha_federatedx.o:ha_federatedx.cc $(CXXCOMPILE) $(LM_CFLAGS) -c $< ha_heap.o:ha_heap.cc === modified file 'mysql-test/mysql-test-run.pl' --- a/mysql-test/mysql-test-run.pl 2009-10-26 11:35:42 +0000 +++ b/mysql-test/mysql-test-run.pl 2009-10-30 18:50:56 +0000 @@ -127,7 +127,6 @@ my $path_config_file; # The ge our $opt_vs_config = $ENV{'MTR_VS_CONFIG'}; my $DEFAULT_SUITES= "binlog,federated,main,maria,rpl,innodb,parts"; -my $opt_suites; our $opt_usage; our $opt_list_options; === modified file 'mysql-test/suite/federated/disabled.def' --- a/mysql-test/suite/federated/disabled.def 2007-12-12 17:19:24 +0000 +++ b/mysql-test/suite/federated/disabled.def 2009-10-30 18:50:56 +0000 @@ -9,4 +9,5 @@ # Do not use any TAB characters for whitespace. # ############################################################################## -federated_transactions : Bug#29523 Transactions do not work +federated_server : needs fixup + === modified file 'mysql-test/suite/federated/federated.result' --- a/mysql-test/suite/federated/federated.result 2009-03-19 08:49:51 +0000 +++ b/mysql-test/suite/federated/federated.result 2009-10-30 18:50:56 +0000 @@ -47,9 +47,10 @@ CREATE TABLE federated.t1 ( ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t3'; -SELECT * FROM federated.t1; -ERROR HY000: The foreign data source you are trying to reference does not exist. Data source error: error: 1146 'Table 'federated.t3' doesn't exist' -DROP TABLE federated.t1; +ERROR HY000: Can't create federated table. Foreign data src error: database: 'federated' username: 'root' hostname: '127.0.0.1' +DROP TABLE IF EXISTS federated.t1; +Warnings: +Note 1051 Unknown table 't1' CREATE TABLE federated.t1 ( `id` int(20) NOT NULL, `group` int NOT NULL default 0, @@ -59,9 +60,10 @@ CREATE TABLE federated.t1 ( ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 CONNECTION='mysql://user:pass@127.0.0.1:SLAVE_PORT/federated/t1'; -SELECT * FROM federated.t1; -ERROR HY000: Unable to connect to foreign data source: Access denied for user 'user'@'localhost' (using password: YES) -DROP TABLE federated.t1; +ERROR HY000: Can't create federated table. Foreign data src error: database: 'federated' username: 'user' hostname: '127.0.0.1' +DROP TABLE IF EXISTS federated.t1; +Warnings: +Note 1051 Unknown table 't1' CREATE TABLE federated.t1 ( `id` int(20) NOT NULL, `group` int NOT NULL default 0, @@ -1944,15 +1946,7 @@ Bug#18287 create federated table always Test that self-references work -create table federated.t1 (a int primary key); -create table federated.t2 (a int primary key) -ENGINE=FEDERATED -connection='mysql://root@127.0.0.1:SLAVE_PORT/federated/t1'; -insert into federated.t1 (a) values (1); -select * from federated.t2; -a -1 -drop table federated.t1, federated.t2; +fix LOCK_open before reenabling test for Bug#18287 CREATE TABLE federated.t1 (a INT PRIMARY KEY) DEFAULT CHARSET=utf8; CREATE TABLE federated.t1 (a INT PRIMARY KEY) ENGINE=FEDERATED @@ -1960,13 +1954,11 @@ CONNECTION='mysql://root@127.0.0.1:SLAVE DEFAULT CHARSET=utf8; SELECT transactions FROM information_schema.engines WHERE engine="FEDERATED"; transactions -NO +YES INSERT INTO federated.t1 VALUES (1); SET autocommit=0; INSERT INTO federated.t1 VALUES (2); ROLLBACK; -Warnings: -Warning 1196 Some non-transactional changed tables couldn't be rolled back SET autocommit=1; SELECT * FROM federated.t1; a @@ -2157,6 +2149,6 @@ End of 5.1 tests SET @@GLOBAL.CONCURRENT_INSERT= @OLD_MASTER_CONCURRENT_INSERT; SET @@GLOBAL.CONCURRENT_INSERT= @OLD_SLAVE_CONCURRENT_INSERT; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; === modified file 'mysql-test/suite/federated/federated.test' --- a/mysql-test/suite/federated/federated.test 2009-03-19 08:49:51 +0000 +++ b/mysql-test/suite/federated/federated.test 2009-10-30 18:50:56 +0000 @@ -57,6 +57,7 @@ CREATE TABLE federated.t1 ( # test non-existant table --replace_result $SLAVE_MYPORT SLAVE_PORT +--error ER_CANT_CREATE_FEDERATED_TABLE eval CREATE TABLE federated.t1 ( `id` int(20) NOT NULL, `group` int NOT NULL default 0, @@ -66,12 +67,11 @@ eval CREATE TABLE federated.t1 ( ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t3'; ---error ER_FOREIGN_DATA_SOURCE_DOESNT_EXIST -SELECT * FROM federated.t1; -DROP TABLE federated.t1; +DROP TABLE IF EXISTS federated.t1; # test bad user/password --replace_result $SLAVE_MYPORT SLAVE_PORT +--error ER_CANT_CREATE_FEDERATED_TABLE eval CREATE TABLE federated.t1 ( `id` int(20) NOT NULL, `group` int NOT NULL default 0, @@ -81,9 +81,7 @@ eval CREATE TABLE federated.t1 ( ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 CONNECTION='mysql://user:pass@127.0.0.1:$SLAVE_MYPORT/federated/t1'; ---error ER_CONNECT_TO_FOREIGN_DATA_SOURCE -SELECT * FROM federated.t1; -DROP TABLE federated.t1; +DROP TABLE IF EXISTS federated.t1; # # correct connection, same named tables --replace_result $SLAVE_MYPORT SLAVE_PORT @@ -1806,6 +1804,8 @@ drop table federated.t1; --echo --echo Test that self-references work --echo +--echo fix LOCK_open before reenabling test for Bug#18287 +--disable_parsing connection slave; create table federated.t1 (a int primary key); --replace_result $SLAVE_MYPORT SLAVE_PORT @@ -1815,7 +1815,7 @@ eval create table federated.t2 (a int pr insert into federated.t1 (a) values (1); select * from federated.t2; drop table federated.t1, federated.t2; - +--enable_parsing # # BUG#29875 Disable support for transactions # === modified file 'mysql-test/suite/federated/federated_archive.result' --- a/mysql-test/suite/federated/federated_archive.result 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_archive.result 2009-10-30 18:50:56 +0000 @@ -34,6 +34,6 @@ id name DROP TABLE federated.t1; DROP TABLE federated.archive_table; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; === modified file 'mysql-test/suite/federated/federated_bug_13118.result' --- a/mysql-test/suite/federated/federated_bug_13118.result 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_bug_13118.result 2009-10-30 18:50:56 +0000 @@ -25,6 +25,6 @@ foo bar DROP TABLE federated.t1; DROP TABLE federated.bug_13118_table; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; === modified file 'mysql-test/suite/federated/federated_bug_25714.result' --- a/mysql-test/suite/federated/federated_bug_25714.result 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_bug_25714.result 2009-10-30 18:50:56 +0000 @@ -48,6 +48,6 @@ SET @@GLOBAL.CONCURRENT_INSERT= @OLD_MAS DROP TABLE federated.t1; SET @@GLOBAL.CONCURRENT_INSERT= @OLD_SLAVE_CONCURRENT_INSERT; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; === modified file 'mysql-test/suite/federated/federated_cleanup.inc' --- a/mysql-test/suite/federated/federated_cleanup.inc 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_cleanup.inc 2009-10-30 18:50:56 +0000 @@ -1,9 +1,9 @@ connection master; --disable_warnings DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; connection slave; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; --enable_warnings === modified file 'mysql-test/suite/federated/federated_innodb.result' --- a/mysql-test/suite/federated/federated_innodb.result 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_innodb.result 2009-10-30 18:50:56 +0000 @@ -20,6 +20,6 @@ a b drop table federated.t1; drop table federated.t1; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; DROP TABLE IF EXISTS federated.t1; -DROP DATABASE federated; +DROP DATABASE IF EXISTS federated; === modified file 'mysql-test/suite/federated/federated_server.result' --- a/mysql-test/suite/federated/federated_server.result 2009-02-02 11:36:03 +0000 +++ b/mysql-test/suite/federated/federated_server.result 2009-10-30 18:50:56 +0000 @@ -178,8 +178,8 @@ INSERT INTO db_bogus.t1 VALUES ('2','thi create server 's1' foreign data wrapper 'mysql' options (HOST '127.0.0.1', DATABASE 'db_legitimate', -USER 'root', -PASSWORD '', +USER 'test_fed', +PASSWORD 'foo', PORT SLAVE_PORT, SOCKET '', OWNER 'root'); === modified file 'mysql-test/suite/federated/federated_server.test' --- a/mysql-test/suite/federated/federated_server.test 2009-06-05 15:35:22 +0000 +++ b/mysql-test/suite/federated/federated_server.test 2009-10-30 18:50:56 +0000 @@ -3,6 +3,7 @@ # Slow test, don't run during staging part -- source include/not_staging.inc +-- source include/big_test.inc -- source federated.inc connection slave; @@ -182,13 +183,17 @@ CREATE TABLE db_bogus.t1 ( ; INSERT INTO db_bogus.t1 VALUES ('2','this is bogus'); +connection slave; +create user test_fed@localhost identified by 'foo'; +grant all on db_legitimate.* to test_fed@localhost; + connection master; --replace_result $SLAVE_MYPORT SLAVE_PORT eval create server 's1' foreign data wrapper 'mysql' options (HOST '127.0.0.1', DATABASE 'db_legitimate', - USER 'root', - PASSWORD '', + USER 'test_fed', + PASSWORD 'foo', PORT $SLAVE_MYPORT, SOCKET '', OWNER 'root'); === modified file 'mysql-test/suite/federated/federated_transactions.result' --- a/mysql-test/suite/federated/federated_transactions.result 2007-12-12 17:19:24 +0000 +++ b/mysql-test/suite/federated/federated_transactions.result 2009-10-30 18:50:56 +0000 @@ -1,13 +1,4 @@ -stop slave; -drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9; -reset master; -reset slave; -drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9; -start slave; -stop slave; -DROP DATABASE IF EXISTS federated; CREATE DATABASE federated; -DROP DATABASE IF EXISTS federated; CREATE DATABASE federated; DROP TABLE IF EXISTS federated.t1; Warnings: === modified file 'mysql-test/valgrind.supp' --- a/mysql-test/valgrind.supp 2009-09-15 10:46:35 +0000 +++ b/mysql-test/valgrind.supp 2009-10-30 18:50:56 +0000 @@ -880,3 +880,17 @@ fun:nptl_pthread_exit_hack_handler fun:start_thread } + +# +# Problem with glibc and gethostbyaddr_r +# + +{ + libc_res_nsend: Conditional jump or move depends on uninitialised value + Memcheck:Cond + fun: __libc_res_nsend + fun: __libc_res_nquery + obj: /lib64/libnss_dns-*so) + obj: /lib64/libnss_dns-*so) + fun: gethostbyaddr_r +} === added file 'storage/federated/README' --- a/storage/federated/README 1970-01-01 00:00:00 +0000 +++ b/storage/federated/README 2009-10-30 18:50:56 +0000 @@ -0,0 +1,7 @@ +The files in this directory are not used by MariaDB + +MariaDB uses the new federated storage engine that can be found in the +federatedx directory. + +This directory is only kept around to make it easy to merge code from the +MySQL source repositories that uses the old and disabled federated code. === renamed file 'storage/federated/plug.in' => 'storage/federated/plug.in.disabled' === added directory 'storage/federatedx' === added file 'storage/federatedx/AUTHORS' --- a/storage/federatedx/AUTHORS 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/AUTHORS 2009-10-30 18:50:56 +0000 @@ -0,0 +1,11 @@ +FederatedX + +Patrick Galbraith <patg(a)patg.net> - Federated + +Pluggable Storage Engine Skeleton setup + +Brian Aker <brian(a)mysql.com> | <brian(a)tangent.org> - Original Design +Calvin Sun - Windows Support +Brian Miezejewski - Bug fixes +Antony T Curtis - Help in inital development, transactions and various help +Michael Widenius - Bug fixes and some simple early optimizations === added file 'storage/federatedx/CMakeFiles.txt' --- a/storage/federatedx/CMakeFiles.txt 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/CMakeFiles.txt 2009-10-30 18:50:56 +0000 @@ -0,0 +1,3 @@ +INCLUDE("${PROJECT_SOURCE_DIR}/storage/mysql_storage_engine.cmake") +SET(FEDERATED_SOURCES ha_federatedx.cc) +MYSQL_STORAGE_ENGINE(FEDERATED) === added file 'storage/federatedx/ChangeLog' --- a/storage/federatedx/ChangeLog 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/ChangeLog 2009-10-30 18:50:56 +0000 @@ -0,0 +1,18 @@ +0.2 - Thu March 8 00:00:00 EST 2008 + + - Fixed bug #30051 "CREATE TABLE does not connect and check existence of remote table" + Modified "real_connect" to take a share and create flag to in order to not rely + on any settings that are later instantiated and/or set by get_share + Also, put logic in the code to not attempt this if a localhost. There's an annoying + functionality that if federated tries to connect to itself during creater table, you + get 1159 error (timeout) - only when local. This prevents having this functionality + and is probably part of the reason it was removed. + +0.1 - Thu Feb 1 00:00:00 EST 2008 + + - This is the FederatedX Storage Engine, + first release. + - Added documentation + - Added simple test and README file to explain + how to run the test + - Added FAQ === added file 'storage/federatedx/FAQ' --- a/storage/federatedx/FAQ 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/FAQ 2009-10-30 18:50:56 +0000 @@ -0,0 +1,40 @@ +Q. What is the FederatedX pluggable storage engine? + +A. It is a fork of the Federated Storage Engine that Brian Aker and I +(Patrick Galbraith) developed originally . It is a storage engine that +uses a client connection to a remote MySQL data source as its data +source instead of a local file on disk. + +Q. Why did you fork from Federated? + +A. To enhance the storage engine independently of the +MySQL Server release schedule. Many people have been +mentioning their dissatisfaction with the limitations +of Federated. I think the engine is a great concept and +have a sense of obligation to continue to improve it. +There are some patches already that are in dire need +of being applied and tested. + +Q. What do you plan to do with FederatedX? + +A. Many things need addressing: + +- Outstanding bugs +- How do deal with huge result sets +- Pushdown conditions (being able to pass things like LIMIT + to the remote connection to keep from returning huge + result sets). +- Better transactional support +- Other connection mechanisms (ODBC, JDBC, native drivers + of other RDBMSs) + +Q. What FederatedX is and is not? + +A. FederatedX is not yet a complete "federated" solution in + the sense that other venders have developed (IBM, etc). It + is essentially a networked storage engine. It is my hope + to make it a real federated solution. + +Q. In which MySQL distributions/forks/branches can I find FederateX + +A. MariaDB (http://www.mariadb.com) === added file 'storage/federatedx/Makefile.am' --- a/storage/federatedx/Makefile.am 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/Makefile.am 2009-10-30 18:50:56 +0000 @@ -0,0 +1,64 @@ +# Used to build Makefile.in + +MYSQLDATAdir = $(localstatedir) +MYSQLSHAREdir = $(pkgdatadir) +MYSQLBASEdir= $(prefix) +MYSQLLIBdir= $(pkglibdir) +pkgplugindir = $(pkglibdir)/plugin +INCLUDES = -I$(top_srcdir)/include -I$(top_builddir)/include \ + -I$(top_srcdir)/regex \ + -I$(top_srcdir)/sql \ + -I$(srcdir) +WRAPLIBS= + +LDADD = + +DEFS = @DEFS@ + +noinst_HEADERS = ha_federatedx.h federatedx_probes.h + +EXTRA_LTLIBRARIES = ha_federatedx.la +pkgplugin_LTLIBRARIES = @plugin_federated_shared_target@ +ha_federatedx_la_LDFLAGS = -module -rpath $(pkgplugindir) +ha_federatedx_la_CXXFLAGS= $(AM_CFLAGS) -DMYSQL_DYNAMIC_PLUGIN +ha_federatedx_la_CFLAGS = $(AM_CFLAGS) -DMYSQL_DYNAMIC_PLUGIN +ha_federatedx_la_SOURCES = ha_federatedx.cc + + +EXTRA_LIBRARIES = libfederatedx.a +noinst_LIBRARIES = @plugin_federated_static_target@ +libfederatedx_a_CXXFLAGS = $(AM_CFLAGS) +libfederatedx_a_CFLAGS = $(AM_CFLAGS) +libfederatedx_a_SOURCES= ha_federatedx.cc federatedx_txn.cc \ + federatedx_io.cc federatedx_io_null.cc \ + federatedx_io_mysql.cc + +EXTRA_DIST = CMakeLists.txt plug.in ha_federatedx.h \ + federatedx_probes.d + +ha_federatedx_la_SOURCES = ha_federatedx.cc federatedx_txn.cc \ + federatedx_io.cc federatedx_io_null.cc \ + federatedx_io_mysql.cc $(top_srcdir)/mysys/string.c +ha_federatedx_la_LIBADD = + +#DTRACE = @DTRACE@ +#DTRACEFLAGS = @DTRACEFLAGS@ +#DTRACEFILES = .libs/libfederatedx_engine_la-ha_federatedx.o + +# #if HAVE_DTRACE +# # libfederatedx_engine_la_LIBADD += federatedx_probes.o +# #endif + +# federatedx_probes.h: federatedx_probes.d +# $(DTRACE) $(DTRACEFLAGS) -h -s federatedx_probes.d +# mv federatedx_probes.h federatedx_probes.h.bak +# sed "s/#include <unistd.h>//g" federatedx_probes.h.bak > federatedx_probes.h +# rm federatedx_probes.h.bak + +#federatedx_probes.o: +# $(DTRACE) $(DTRACEFLAGS) -G -s federatedx_probes.d $(DTRACEFILES) + +# End + +# Don't update the files from bitkeeper +%::SCCS/s.% === added file 'storage/federatedx/README' --- a/storage/federatedx/README 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/README 2009-10-30 18:50:56 +0000 @@ -0,0 +1,33 @@ +This is the FederatedX Storage Engine, developed as an external storage engine. + +NOTE: + +The following is only relevant if you use it for MySQL. MariaDB already comes +with the latest version of FederatedX. + +To install, grab a copy of the mysql source code and run this: + +./configure --with-mysql=/path/to/src/mysql-5.x --libdir=/usr/local/lib/mysql/ + +make install + +And then inside of MySQL: + +mysql> INSTALL PLUGIN federatedx SONAME 'libfederatedx_engine.so'; + +mysql> CREATE TABLE `d` (`a` varchar(125), b text, primary key(a)) ENGINE=FEDERATEDX CONNECTION="mysql://root@host/schema/table" + +or + +mysql> CREATE TABLE `d` (`a` varchar(125), b text, primary key(a)) ENGINE=FEDERATEDX CONNECTION="server" CHARSET=latin1; + +You will probably need to edit the Makefile.am in the src/ tree if you want +to build on anything other then Linux (and the Makefile assumes that the +server was not compiled for debug). The reason for the two possible +configure lines is that libdir is dependent on where MySQL was installed. If +you run the "INSTALL PLUGIN ..." and you get a file not found, check that +your configured this directory correctly. + +For Solaris you can enable DTrace probes by adding to configure +--enable-dtrace + === added file 'storage/federatedx/README.windows' --- a/storage/federatedx/README.windows 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/README.windows 2009-10-30 18:50:56 +0000 @@ -0,0 +1,23 @@ +The following files are changed in order to build a new engine on Windows: + +- Update win\configure.js with +case "WITH_FEDERATEDX_STORAGE_ENGINE": +to make sure it will pass WITH_FEDERATEDX_STORAGE_ENGINE in. + +- Update CMakeFiles.txt under mysql root: + IF(WITH_FEDERATEDX_STORAGE_ENGINE) + ADD_DEFINITIONS(-D WITH_FEDERATEDX_STORAGE_ENGINE) + SET (mysql_plugin_defs + "${mysql_plugin_defs},builtin_skeleton_plugin") + ENDIF(WITH_FEDERATEDX_STORAGE_ENGINE) + + and, + + IF(WITH_FEDERATEDX_STORAGE_ENGINE) + ADD_SUBDIRECTORY(storage/skeleton/src) + ENDIF(WITH_FEDERATEDX_STORAGE_ENGINE) + + - Update CMakeFiles.txt under sql: + IF(WITH_FEDERATEDX_STORAGE_ENGINE) + TARGET_LINK_LIBRARIES(mysqld skeleton) + ENDIF(WITH_FEDERATEDX_STORAGE_ENGINE) === added file 'storage/federatedx/TODO' --- a/storage/federatedx/TODO 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/TODO 2009-10-30 18:50:56 +0000 @@ -0,0 +1,30 @@ +Features + +* Add Pushdown conditions +* Add other network driver interfaces +* Handle large result sets +* Auto-discovery of tables on foreign data sources + +Bugs (http://bugs.mysql.com) + +20026 2006-05-23 FEDERATED lacks support for auto_increment_increment and auto_increment_offset +20724 2006-06-27 FEDERATED does not honour SET INSERT_ID +28269 2007-05-06 Any FEDERATED engine fails to quote reserved words for field names +25509 2007-01-10 Federated: Failure with non-ASCII characters +26697 2007-02-27 Every query to a federated table results in a full scan of MyISAM table. +21360 2006-07-31 Microsoft Windows (Windows/Linux) mysqldump error on federated tables +34189 2008-01-31 Any ALTER TABLE t1 ENGINE=FEDERATED CONNECTION='connectionString' on MyISAM fails +31757 2007-10-22 Any Federated tables break replication Antony Curtis +33953 2008-01-21 Any mysqld dies on search federated table using nullable index with < or <= operator +34015 2008-01-23 Linux Problems with float fields using federated tables +21583 2006-08-11 Linux (Linux) Federated table returns broken strings. +33702 2008-01-05 Accessing a federated table with a non existing server returns random error code +25512 2007-01-10 Federated: CREATE failures +32426 2007-11-16 Any FEDERATED query returns corrupt results for ORDER BY on a TEXT field +25510 2007-01-10 Federated: double trigger activation +33250 2007-12-14 SELECT * FROM really_big_federated_table eats lots of virtual memory (OOM) +14874 2005-11-11 Error 2013: Lost connection to MySQL server with Federated table +25508 2007-01-10 Federated: Failure to Remove Partitioning +27180 2007-03-15 #1030 - Got error 1 from storage engine with big tables +33947 2008-01-20 Any Join on Federated tables with Unique index and IS NOT NULL crashes server +30051 (fixed) CREATE TABLE does not connect and check existence of remote table === added file 'storage/federatedx/federatedx_io.cc' --- a/storage/federatedx/federatedx_io.cc 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/federatedx_io.cc 2009-10-30 18:50:56 +0000 @@ -0,0 +1,103 @@ +/* +Copyright (c) 2007, Antony T Curtis +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Neither the name of FederatedX nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + + +/*#define MYSQL_SERVER 1*/ +#include "mysql_priv.h" +#include <mysql/plugin.h> + +#include "ha_federatedx.h" + +#include "m_string.h" + +#ifdef USE_PRAGMA_IMPLEMENTATION +#pragma implementation // gcc: Class implementation +#endif + +typedef federatedx_io *(*instantiate_io_type)(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server); +struct io_schemes_st +{ + const char *scheme; + instantiate_io_type instantiate; +}; + + +static const io_schemes_st federated_io_schemes[] = +{ + { "mysql", &instantiate_io_mysql }, + { "null", instantiate_io_null } /* must be last element */ +}; + +const uint federated_io_schemes_count= array_elements(federated_io_schemes); + +federatedx_io::federatedx_io(FEDERATEDX_SERVER *aserver) + : server(aserver), owner_ptr(0), txn_next(0), idle_next(0), + active(FALSE), busy(FALSE), readonly(TRUE) +{ + DBUG_ENTER("federatedx_io::federatedx_io"); + DBUG_ASSERT(server); + + safe_mutex_assert_owner(&server->mutex); + server->io_count++; + + DBUG_VOID_RETURN; +} + + +federatedx_io::~federatedx_io() +{ + DBUG_ENTER("federatedx_io::~federatedx_io"); + + server->io_count--; + + DBUG_VOID_RETURN; +} + + +bool federatedx_io::handles_scheme(const char *scheme) +{ + const io_schemes_st *ptr = federated_io_schemes; + const io_schemes_st *end = ptr + array_elements(federated_io_schemes); + while (ptr != end && strcasecmp(scheme, ptr->scheme)) + ++ptr; + return ptr != end; +} + + +federatedx_io *federatedx_io::construct(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server) +{ + const io_schemes_st *ptr = federated_io_schemes; + const io_schemes_st *end = ptr + (array_elements(federated_io_schemes) - 1); + while (ptr != end && strcasecmp(server->scheme, ptr->scheme)) + ++ptr; + return ptr->instantiate(server_root, server); +} + + === added file 'storage/federatedx/federatedx_io_mysql.cc' --- a/storage/federatedx/federatedx_io_mysql.cc 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/federatedx_io_mysql.cc 2009-10-30 18:50:56 +0000 @@ -0,0 +1,592 @@ +/* +Copyright (c) 2007, Antony T Curtis +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Neither the name of FederatedX nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + + +/*#define MYSQL_SERVER 1*/ +#include "mysql_priv.h" +#include <mysql/plugin.h> + +#include "ha_federatedx.h" + +#include "m_string.h" + +#ifdef USE_PRAGMA_IMPLEMENTATION +#pragma implementation // gcc: Class implementation +#endif + + +#define SAVEPOINT_REALIZED 1 +#define SAVEPOINT_RESTRICT 2 +#define SAVEPOINT_EMITTED 4 + + +typedef struct federatedx_savepoint +{ + ulong level; + uint flags; +} SAVEPT; + + +class federatedx_io_mysql :public federatedx_io +{ + MYSQL mysql; /* MySQL connection */ + DYNAMIC_ARRAY savepoints; + bool requested_autocommit; + bool actual_autocommit; + + int actual_query(const char *buffer, uint length); + bool test_all_restrict() const; +public: + federatedx_io_mysql(FEDERATEDX_SERVER *); + ~federatedx_io_mysql(); + + int simple_query(const char *fmt, ...); + int query(const char *buffer, uint length); + virtual FEDERATEDX_IO_RESULT *store_result(); + + virtual size_t max_query_size() const; + + virtual my_ulonglong affected_rows() const; + virtual my_ulonglong last_insert_id() const; + + virtual int error_code(); + virtual const char *error_str(); + + void reset(); + int commit(); + int rollback(); + + int savepoint_set(ulong sp); + ulong savepoint_release(ulong sp); + ulong savepoint_rollback(ulong sp); + void savepoint_restrict(ulong sp); + + ulong last_savepoint() const; + ulong actual_savepoint() const; + bool is_autocommit() const; + + bool table_metadata(ha_statistics *stats, const char *table_name, + uint table_name_length, uint flag); + + /* resultset operations */ + + virtual void free_result(FEDERATEDX_IO_RESULT *io_result); + virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result); + virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result); + virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result); + virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result); + virtual const char *get_column_data(FEDERATEDX_IO_ROW *row, + unsigned int column); + virtual bool is_column_null(const FEDERATEDX_IO_ROW *row, + unsigned int column) const; +}; + + +federatedx_io *instantiate_io_mysql(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server) +{ + return new (server_root) federatedx_io_mysql(server); +} + + +federatedx_io_mysql::federatedx_io_mysql(FEDERATEDX_SERVER *aserver) + : federatedx_io(aserver), + requested_autocommit(TRUE), actual_autocommit(TRUE) +{ + DBUG_ENTER("federatedx_io_mysql::federatedx_io_mysql"); + + bzero(&mysql, sizeof(MYSQL)); + bzero(&savepoints, sizeof(DYNAMIC_ARRAY)); + + my_init_dynamic_array(&savepoints, sizeof(SAVEPT), 16, 16); + + DBUG_VOID_RETURN; +} + + +federatedx_io_mysql::~federatedx_io_mysql() +{ + DBUG_ENTER("federatedx_io_mysql::~federatedx_io_mysql"); + + mysql_close(&mysql); + delete_dynamic(&savepoints); + + DBUG_VOID_RETURN; +} + + +void federatedx_io_mysql::reset() +{ + reset_dynamic(&savepoints); + set_active(FALSE); + + requested_autocommit= TRUE; + mysql.reconnect= 1; +} + + +int federatedx_io_mysql::commit() +{ + int error= 0; + DBUG_ENTER("federatedx_io_mysql::commit"); + + if (!actual_autocommit && (error= actual_query("COMMIT", 6))) + rollback(); + + reset(); + + DBUG_RETURN(error); +} + +int federatedx_io_mysql::rollback() +{ + int error= 0; + DBUG_ENTER("federatedx_io_mysql::rollback"); + + if (!actual_autocommit) + error= actual_query("ROLLBACK", 8); + else + error= ER_WARNING_NOT_COMPLETE_ROLLBACK; + + reset(); + + DBUG_RETURN(error); +} + + +ulong federatedx_io_mysql::last_savepoint() const +{ + SAVEPT *savept= NULL; + DBUG_ENTER("federatedx_io_mysql::last_savepoint"); + + if (savepoints.elements) + savept= dynamic_element(&savepoints, savepoints.elements - 1, SAVEPT *); + + DBUG_RETURN(savept ? savept->level : 0); +} + + +ulong federatedx_io_mysql::actual_savepoint() const +{ + SAVEPT *savept= NULL; + uint index= savepoints.elements; + DBUG_ENTER("federatedx_io_mysql::last_savepoint"); + + while (index) + { + savept= dynamic_element(&savepoints, --index, SAVEPT *); + if (savept->flags & SAVEPOINT_REALIZED) + break; + savept= NULL; + } + + DBUG_RETURN(savept ? savept->level : 0); +} + +bool federatedx_io_mysql::is_autocommit() const +{ + return actual_autocommit; +} + + +int federatedx_io_mysql::savepoint_set(ulong sp) +{ + int error; + SAVEPT savept; + DBUG_ENTER("federatedx_io_mysql::savepoint_set"); + DBUG_PRINT("info",("savepoint=%lu", sp)); + DBUG_ASSERT(sp > last_savepoint()); + + savept.level= sp; + savept.flags= 0; + + if ((error= insert_dynamic(&savepoints, (uchar*) &savept) ? -1 : 0)) + goto err; + + set_active(TRUE); + mysql.reconnect= 0; + requested_autocommit= FALSE; + +err: + DBUG_RETURN(error); +} + + +ulong federatedx_io_mysql::savepoint_release(ulong sp) +{ + SAVEPT *savept, *last= NULL; + DBUG_ENTER("federatedx_io_mysql::savepoint_release"); + DBUG_PRINT("info",("savepoint=%lu", sp)); + + while (savepoints.elements) + { + savept= dynamic_element(&savepoints, savepoints.elements - 1, SAVEPT *); + if (savept->level < sp) + break; + if ((savept->flags & (SAVEPOINT_REALIZED | + SAVEPOINT_RESTRICT)) == SAVEPOINT_REALIZED) + last= savept; + savepoints.elements--; + } + + if (last) + { + char buffer[STRING_BUFFER_USUAL_SIZE]; + int length= my_snprintf(buffer, sizeof(buffer), + "RELEASE SAVEPOINT save%lu", last->level); + actual_query(buffer, length); + } + + DBUG_RETURN(last_savepoint()); +} + + +ulong federatedx_io_mysql::savepoint_rollback(ulong sp) +{ + SAVEPT *savept; + uint index; + DBUG_ENTER("federatedx_io_mysql::savepoint_release"); + DBUG_PRINT("info",("savepoint=%lu", sp)); + + while (savepoints.elements) + { + savept= dynamic_element(&savepoints, savepoints.elements - 1, SAVEPT *); + if (savept->level <= sp) + break; + savepoints.elements--; + } + + for (index= savepoints.elements, savept= NULL; index;) + { + savept= dynamic_element(&savepoints, --index, SAVEPT *); + if (savept->flags & SAVEPOINT_REALIZED) + break; + savept= NULL; + } + + if (savept && !(savept->flags & SAVEPOINT_RESTRICT)) + { + char buffer[STRING_BUFFER_USUAL_SIZE]; + int length= my_snprintf(buffer, sizeof(buffer), + "ROLLBACK TO SAVEPOINT save%lu", savept->level); + actual_query(buffer, length); + } + + DBUG_RETURN(last_savepoint()); +} + + +void federatedx_io_mysql::savepoint_restrict(ulong sp) +{ + SAVEPT *savept; + uint index= savepoints.elements; + DBUG_ENTER("federatedx_io_mysql::savepoint_restrict"); + + while (index) + { + savept= dynamic_element(&savepoints, --index, SAVEPT *); + if (savept->level > sp) + continue; + if (savept->level < sp) + break; + savept->flags|= SAVEPOINT_RESTRICT; + break; + } + + DBUG_VOID_RETURN; +} + + +int federatedx_io_mysql::simple_query(const char *fmt, ...) +{ + char buffer[STRING_BUFFER_USUAL_SIZE]; + int length, error; + va_list arg; + DBUG_ENTER("federatedx_io_mysql::simple_query"); + + va_start(arg, fmt); + length= my_vsnprintf(buffer, sizeof(buffer), fmt, arg); + va_end(arg); + + error= query(buffer, length); + + DBUG_RETURN(error); +} + + +bool federatedx_io_mysql::test_all_restrict() const +{ + bool result= FALSE; + SAVEPT *savept; + uint index= savepoints.elements; + DBUG_ENTER("federatedx_io_mysql::test_all_restrict"); + + while (index) + { + savept= dynamic_element(&savepoints, --index, SAVEPT *); + if ((savept->flags & (SAVEPOINT_REALIZED | + SAVEPOINT_RESTRICT)) == SAVEPOINT_REALIZED || + (savept->flags & SAVEPOINT_EMITTED)) + DBUG_RETURN(FALSE); + if (savept->flags & SAVEPOINT_RESTRICT) + result= TRUE; + } + + DBUG_RETURN(result); +} + + +int federatedx_io_mysql::query(const char *buffer, uint length) +{ + int error; + bool wants_autocommit= requested_autocommit | is_readonly(); + DBUG_ENTER("federatedx_io_mysql::query"); + + if (!wants_autocommit && test_all_restrict()) + wants_autocommit= TRUE; + + if (wants_autocommit != actual_autocommit) + { + if ((error= actual_query(wants_autocommit ? "SET AUTOCOMMIT=1" + : "SET AUTOCOMMIT=0", 16))) + DBUG_RETURN(error); + mysql.reconnect= wants_autocommit ? 1 : 0; + actual_autocommit= wants_autocommit; + } + + if (!actual_autocommit && last_savepoint() != actual_savepoint()) + { + SAVEPT *savept= dynamic_element(&savepoints, savepoints.elements - 1, + SAVEPT *); + if (!(savept->flags & SAVEPOINT_RESTRICT)) + { + char buf[STRING_BUFFER_USUAL_SIZE]; + int len= my_snprintf(buf, sizeof(buf), + "SAVEPOINT save%lu", savept->level); + if ((error= actual_query(buf, len))) + DBUG_RETURN(error); + set_active(TRUE); + savept->flags|= SAVEPOINT_EMITTED; + } + savept->flags|= SAVEPOINT_REALIZED; + } + + if (!(error= actual_query(buffer, length))) + set_active(is_active() || !actual_autocommit); + + DBUG_RETURN(error); +} + + +int federatedx_io_mysql::actual_query(const char *buffer, uint length) +{ + int error; + DBUG_ENTER("federatedx_io_mysql::actual_query"); + + if (!mysql.master) + { + if (!(mysql_init(&mysql))) + DBUG_RETURN(-1); + + /* + BUG# 17044 Federated Storage Engine is not UTF8 clean + Add set names to whatever charset the table is at open + of table + */ + /* this sets the csname like 'set names utf8' */ + mysql_options(&mysql, MYSQL_SET_CHARSET_NAME, get_charsetname()); + + if (!mysql_real_connect(&mysql, + get_hostname(), + get_username(), + get_password(), + get_database(), + get_port(), + get_socket(), 0)) + DBUG_RETURN(ER_CONNECT_TO_FOREIGN_DATA_SOURCE); + mysql.reconnect= 1; + } + + error= mysql_real_query(&mysql, buffer, length); + + DBUG_RETURN(error); +} + +size_t federatedx_io_mysql::max_query_size() const +{ + return mysql.net.max_packet_size; +} + + +my_ulonglong federatedx_io_mysql::affected_rows() const +{ + return mysql.affected_rows; +} + + +my_ulonglong federatedx_io_mysql::last_insert_id() const +{ + return mysql.last_used_con->insert_id; +} + + +int federatedx_io_mysql::error_code() +{ + return mysql_errno(&mysql); +} + + +const char *federatedx_io_mysql::error_str() +{ + return mysql_error(&mysql); +} + + +FEDERATEDX_IO_RESULT *federatedx_io_mysql::store_result() +{ + FEDERATEDX_IO_RESULT *result; + DBUG_ENTER("federatedx_io_mysql::store_result"); + + result= (FEDERATEDX_IO_RESULT *) mysql_store_result(&mysql); + + DBUG_RETURN(result); +} + + +void federatedx_io_mysql::free_result(FEDERATEDX_IO_RESULT *io_result) +{ + mysql_free_result((MYSQL_RES *) io_result); +} + + +unsigned int federatedx_io_mysql::get_num_fields(FEDERATEDX_IO_RESULT *io_result) +{ + return mysql_num_fields((MYSQL_RES *) io_result); +} + + +my_ulonglong federatedx_io_mysql::get_num_rows(FEDERATEDX_IO_RESULT *io_result) +{ + return mysql_num_rows((MYSQL_RES *) io_result); +} + + +FEDERATEDX_IO_ROW *federatedx_io_mysql::fetch_row(FEDERATEDX_IO_RESULT *io_result) +{ + return (FEDERATEDX_IO_ROW *) mysql_fetch_row((MYSQL_RES *) io_result); +} + + +ulong *federatedx_io_mysql::fetch_lengths(FEDERATEDX_IO_RESULT *io_result) +{ + return mysql_fetch_lengths((MYSQL_RES *) io_result); +} + + +const char *federatedx_io_mysql::get_column_data(FEDERATEDX_IO_ROW *row, + unsigned int column) +{ + return ((MYSQL_ROW)row)[column]; +} + + +bool federatedx_io_mysql::is_column_null(const FEDERATEDX_IO_ROW *row, + unsigned int column) const +{ + return !((MYSQL_ROW)row)[column]; +} + +bool federatedx_io_mysql::table_metadata(ha_statistics *stats, + const char *table_name, + uint table_name_length, uint flag) +{ + char status_buf[FEDERATEDX_QUERY_BUFFER_SIZE]; + FEDERATEDX_IO_RESULT *result= 0; + FEDERATEDX_IO_ROW *row; + String status_query_string(status_buf, sizeof(status_buf), &my_charset_bin); + int error; + + status_query_string.length(0); + status_query_string.append(STRING_WITH_LEN("SHOW TABLE STATUS LIKE ")); + append_ident(&status_query_string, table_name, + table_name_length, value_quote_char); + + if (query(status_query_string.ptr(), status_query_string.length())) + goto error; + + status_query_string.length(0); + + result= store_result(); + + /* + We're going to use fields num. 4, 12 and 13 of the resultset, + so make sure we have these fields. + */ + if (!result || (get_num_fields(result) < 14)) + goto error; + + if (!get_num_rows(result)) + goto error; + + if (!(row= fetch_row(result))) + goto error; + + /* + deleted is set in ha_federatedx::info + */ + /* + need to figure out what this means as far as federatedx is concerned, + since we don't have a "file" + + data_file_length = ? + index_file_length = ? + delete_length = ? + */ + if (!is_column_null(row, 4)) + stats->records= (ha_rows) my_strtoll10(get_column_data(row, 4), + (char**) 0, &error); + if (!is_column_null(row, 5)) + stats->mean_rec_length= (ulong) my_strtoll10(get_column_data(row, 5), + (char**) 0, &error); + + stats->data_file_length= stats->records * stats->mean_rec_length; + + if (!is_column_null(row, 12)) + stats->update_time= (time_t) my_strtoll10(get_column_data(row, 12), + (char**) 0, &error); + if (!is_column_null(row, 13)) + stats->check_time= (time_t) my_strtoll10(get_column_data(row, 13), + (char**) 0, &error); + + free_result(result); + return 0; + +error: + free_result(result); + return 1; +} === added file 'storage/federatedx/federatedx_io_null.cc' --- a/storage/federatedx/federatedx_io_null.cc 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/federatedx_io_null.cc 2009-10-30 18:50:56 +0000 @@ -0,0 +1,277 @@ +/* +Copyright (c) 2007, Antony T Curtis +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Neither the name of FederatedX nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + + +/*#define MYSQL_SERVER 1*/ +#include "mysql_priv.h" +#include <mysql/plugin.h> + +#include "ha_federatedx.h" + +#include "m_string.h" + +#ifdef USE_PRAGMA_IMPLEMENTATION +#pragma implementation // gcc: Class implementation +#endif + + +#define SAVEPOINT_REALIZED 1 +#define SAVEPOINT_RESTRICT 2 +#define SAVEPOINT_EMITTED 4 + + +typedef struct federatedx_savepoint +{ + ulong level; + uint flags; +} SAVEPT; + + +class federatedx_io_null :public federatedx_io +{ +public: + federatedx_io_null(FEDERATEDX_SERVER *); + ~federatedx_io_null(); + + int query(const char *buffer, uint length); + virtual FEDERATEDX_IO_RESULT *store_result(); + + virtual size_t max_query_size() const; + + virtual my_ulonglong affected_rows() const; + virtual my_ulonglong last_insert_id() const; + + virtual int error_code(); + virtual const char *error_str(); + + void reset(); + int commit(); + int rollback(); + + int savepoint_set(ulong sp); + ulong savepoint_release(ulong sp); + ulong savepoint_rollback(ulong sp); + void savepoint_restrict(ulong sp); + + ulong last_savepoint() const; + ulong actual_savepoint() const; + bool is_autocommit() const; + + bool table_metadata(ha_statistics *stats, const char *table_name, + uint table_name_length, uint flag); + + /* resultset operations */ + + virtual void free_result(FEDERATEDX_IO_RESULT *io_result); + virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result); + virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result); + virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result); + virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result); + virtual const char *get_column_data(FEDERATEDX_IO_ROW *row, + unsigned int column); + virtual bool is_column_null(const FEDERATEDX_IO_ROW *row, + unsigned int column) const; +}; + + +federatedx_io *instantiate_io_null(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server) +{ + return new (server_root) federatedx_io_null(server); +} + + +federatedx_io_null::federatedx_io_null(FEDERATEDX_SERVER *aserver) + : federatedx_io(aserver) +{ +} + + +federatedx_io_null::~federatedx_io_null() +{ +} + + +void federatedx_io_null::reset() +{ +} + + +int federatedx_io_null::commit() +{ + return 0; +} + +int federatedx_io_null::rollback() +{ + return 0; +} + + +ulong federatedx_io_null::last_savepoint() const +{ + return 0; +} + + +ulong federatedx_io_null::actual_savepoint() const +{ + return 0; +} + +bool federatedx_io_null::is_autocommit() const +{ + return 0; +} + + +int federatedx_io_null::savepoint_set(ulong sp) +{ + return 0; +} + + +ulong federatedx_io_null::savepoint_release(ulong sp) +{ + return 0; +} + + +ulong federatedx_io_null::savepoint_rollback(ulong sp) +{ + return 0; +} + + +void federatedx_io_null::savepoint_restrict(ulong sp) +{ +} + + +int federatedx_io_null::query(const char *buffer, uint length) +{ + return 0; +} + + +size_t federatedx_io_null::max_query_size() const +{ + return INT_MAX; +} + + +my_ulonglong federatedx_io_null::affected_rows() const +{ + return 0; +} + + +my_ulonglong federatedx_io_null::last_insert_id() const +{ + return 0; +} + + +int federatedx_io_null::error_code() +{ + return 0; +} + + +const char *federatedx_io_null::error_str() +{ + return ""; +} + + +FEDERATEDX_IO_RESULT *federatedx_io_null::store_result() +{ + FEDERATEDX_IO_RESULT *result; + DBUG_ENTER("federatedx_io_null::store_result"); + + result= NULL; + + DBUG_RETURN(result); +} + + +void federatedx_io_null::free_result(FEDERATEDX_IO_RESULT *) +{ +} + + +unsigned int federatedx_io_null::get_num_fields(FEDERATEDX_IO_RESULT *) +{ + return 0; +} + + +my_ulonglong federatedx_io_null::get_num_rows(FEDERATEDX_IO_RESULT *) +{ + return 0; +} + + +FEDERATEDX_IO_ROW *federatedx_io_null::fetch_row(FEDERATEDX_IO_RESULT *) +{ + return NULL; +} + + +ulong *federatedx_io_null::fetch_lengths(FEDERATEDX_IO_RESULT *) +{ + return NULL; +} + + +const char *federatedx_io_null::get_column_data(FEDERATEDX_IO_ROW *, + unsigned int) +{ + return ""; +} + + +bool federatedx_io_null::is_column_null(const FEDERATEDX_IO_ROW *, + unsigned int) const +{ + return true; +} + +bool federatedx_io_null::table_metadata(ha_statistics *stats, + const char *table_name, + uint table_name_length, uint flag) +{ + stats->records= (ha_rows) 0; + stats->mean_rec_length= (ulong) 0; + stats->data_file_length= 0; + + stats->update_time= (time_t) 0; + stats->check_time= (time_t) 0; + + return 0; +} === added file 'storage/federatedx/federatedx_probes.h' --- a/storage/federatedx/federatedx_probes.h 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/federatedx_probes.h 2009-10-30 18:50:56 +0000 @@ -0,0 +1,45 @@ +/* + * Generated by dtrace(1M). + */ + +#ifndef _FEDERATED_PROBES_H +#define _FEDERATED_PROBES_H + + + +#ifdef __cplusplus +extern "C" { +#endif + +#if _DTRACE_VERSION + +#define FEDERATED_CLOSE() \ + __dtrace_federated___close() +#define FEDERATED_CLOSE_ENABLED() \ + __dtraceenabled_federated___close() +#define FEDERATED_OPEN() \ + __dtrace_federated___open() +#define FEDERATED_OPEN_ENABLED() \ + __dtraceenabled_federated___open() + + +extern void __dtrace_federated___close(void); +extern int __dtraceenabled_federated___close(void); +extern void __dtrace_federated___open(void); +extern int __dtraceenabled_federated___open(void); + +#else + +#define FEDERATED_CLOSE() +#define FEDERATED_CLOSE_ENABLED() (0) +#define FEDERATED_OPEN() +#define FEDERATED_OPEN_ENABLED() (0) + +#endif + + +#ifdef __cplusplus +} +#endif + +#endif /* _FEDERATED_PROBES_H */ === added file 'storage/federatedx/federatedx_txn.cc' --- a/storage/federatedx/federatedx_txn.cc 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/federatedx_txn.cc 2009-10-30 18:50:56 +0000 @@ -0,0 +1,424 @@ +/* +Copyright (c) 2007, Antony T Curtis +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Neither the name of FederatedX nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + + +/*#define MYSQL_SERVER 1*/ +#include "mysql_priv.h" +#include <mysql/plugin.h> + +#include "ha_federatedx.h" + +#include "m_string.h" + +#ifdef USE_PRAGMA_IMPLEMENTATION +#pragma implementation // gcc: Class implementation +#endif + + +federatedx_txn::federatedx_txn() + : txn_list(0), savepoint_level(0), savepoint_stmt(0), savepoint_next(0) +{ + DBUG_ENTER("federatedx_txn::federatedx_txn"); + DBUG_VOID_RETURN; +} + +federatedx_txn::~federatedx_txn() +{ + DBUG_ENTER("federatedx_txn::~federatedx_txn"); + DBUG_ASSERT(!txn_list); + DBUG_VOID_RETURN; +} + + +void federatedx_txn::close(FEDERATEDX_SERVER *server) +{ + uint count= 0; + federatedx_io *io, **iop; + DBUG_ENTER("federatedx_txn::close"); + + DBUG_ASSERT(!server->use_count); + DBUG_PRINT("info",("use count: %u connections: %u", + server->use_count, server->io_count)); + + for (iop= &txn_list; (io= *iop);) + { + if (io->server != server) + iop= &io->txn_next; + else + { + *iop= io->txn_next; + io->txn_next= NULL; + io->busy= FALSE; + + io->idle_next= server->idle_list; + server->idle_list= io; + } + } + + while ((io= server->idle_list)) + { + server->idle_list= io->idle_next; + delete io; + count++; + } + + DBUG_PRINT("info",("closed %u connections, txn_list: %s", count, + txn_list ? "active": "empty")); + DBUG_VOID_RETURN; +} + + +int federatedx_txn::acquire(FEDERATEDX_SHARE *share, bool readonly, + federatedx_io **ioptr) +{ + federatedx_io *io; + FEDERATEDX_SERVER *server= share->s; + DBUG_ENTER("federatedx_txn::acquire"); + DBUG_ASSERT(ioptr && server); + + if (!(io= *ioptr)) + { + /* check to see if we have an available IO connection */ + for (io= txn_list; io; io= io->txn_next) + if (io->server == server) + break; + + if (!io) + { + /* check to see if there are any unowned IO connections */ + pthread_mutex_lock(&server->mutex); + if ((io= server->idle_list)) + { + server->idle_list= io->idle_next; + io->idle_next= NULL; + } + else + io= federatedx_io::construct(&server->mem_root, server); + + io->txn_next= txn_list; + txn_list= io; + + pthread_mutex_unlock(&server->mutex); + } + + if (io->busy) + *io->owner_ptr= NULL; + + io->busy= TRUE; + io->owner_ptr= ioptr; + } + + DBUG_ASSERT(io->busy && io->server == server); + + io->readonly&= readonly; + + DBUG_RETURN((*ioptr= io) ? 0 : -1); +} + + +void federatedx_txn::release(federatedx_io **ioptr) +{ + federatedx_io *io; + DBUG_ENTER("federatedx_txn::release"); + DBUG_ASSERT(ioptr); + + if ((io= *ioptr)) + { + /* mark as available for reuse in this transaction */ + io->busy= FALSE; + *ioptr= NULL; + + DBUG_PRINT("info", ("active: %d autocommit: %d", + io->active, io->is_autocommit())); + + if (io->is_autocommit()) + io->active= FALSE; + } + + release_scan(); + + DBUG_VOID_RETURN; +} + + +void federatedx_txn::release_scan() +{ + uint count= 0, returned= 0; + federatedx_io *io, **pio; + DBUG_ENTER("federatedx_txn::release_scan"); + + /* return any inactive and idle connections to the server */ + for (pio= &txn_list; (io= *pio); count++) + { + if (io->active || io->busy) + pio= &io->txn_next; + else + { + FEDERATEDX_SERVER *server= io->server; + + /* unlink from list of connections bound to the transaction */ + *pio= io->txn_next; + io->txn_next= NULL; + + /* reset some values */ + io->readonly= TRUE; + + pthread_mutex_lock(&server->mutex); + io->idle_next= server->idle_list; + server->idle_list= io; + pthread_mutex_unlock(&server->mutex); + returned++; + } + } + DBUG_PRINT("info",("returned %u of %u connections(s)", returned, count)); + + DBUG_VOID_RETURN; +} + + +bool federatedx_txn::txn_begin() +{ + ulong level= 0; + DBUG_ENTER("federatedx_txn::txn_begin"); + + if (savepoint_next == 0) + { + savepoint_next++; + savepoint_level= savepoint_stmt= 0; + sp_acquire(&level); + } + + DBUG_RETURN(level == 1); +} + + +int federatedx_txn::txn_commit() +{ + int error= 0; + federatedx_io *io; + DBUG_ENTER("federatedx_txn::txn_commit"); + + if (savepoint_next) + { + DBUG_ASSERT(savepoint_stmt != 1); + + for (io= txn_list; io; io= io->txn_next) + { + int rc= 0; + + if (io->active) + rc= io->commit(); + else + io->rollback(); + + if (io->active && rc) + error= -1; + + io->reset(); + } + + release_scan(); + + savepoint_next= savepoint_stmt= savepoint_level= 0; + } + + DBUG_RETURN(error); +} + + +int federatedx_txn::txn_rollback() +{ + int error= 0; + federatedx_io *io; + DBUG_ENTER("federatedx_txn::txn_commit"); + + if (savepoint_next) + { + DBUG_ASSERT(savepoint_stmt != 1); + + for (io= txn_list; io; io= io->txn_next) + { + int rc= io->rollback(); + + if (io->active && rc) + error= -1; + + io->reset(); + } + + release_scan(); + + savepoint_next= savepoint_stmt= savepoint_level= 0; + } + + DBUG_RETURN(error); +} + + +bool federatedx_txn::sp_acquire(ulong *sp) +{ + bool rc= FALSE; + federatedx_io *io; + DBUG_ENTER("federatedx_txn::sp_acquire"); + DBUG_ASSERT(sp && savepoint_next); + + *sp= savepoint_level= savepoint_next++; + + for (io= txn_list; io; io= io->txn_next) + { + if (io->readonly) + continue; + + io->savepoint_set(savepoint_level); + rc= TRUE; + } + + DBUG_RETURN(rc); +} + + +int federatedx_txn::sp_rollback(ulong *sp) +{ + ulong level, new_level= savepoint_level; + federatedx_io *io; + DBUG_ENTER("federatedx_txn::sp_rollback"); + DBUG_ASSERT(sp && savepoint_next && *sp && *sp <= savepoint_level); + + for (io= txn_list; io; io= io->txn_next) + { + if (io->readonly) + continue; + + if ((level= io->savepoint_rollback(*sp)) < new_level) + new_level= level; + } + + savepoint_level= new_level; + + DBUG_RETURN(0); +} + + +int federatedx_txn::sp_release(ulong *sp) +{ + ulong level, new_level= savepoint_level; + federatedx_io *io; + DBUG_ENTER("federatedx_txn::sp_release"); + DBUG_ASSERT(sp && savepoint_next && *sp && *sp <= savepoint_level); + + for (io= txn_list; io; io= io->txn_next) + { + if (io->readonly) + continue; + + if ((level= io->savepoint_release(*sp)) < new_level) + new_level= level; + } + + savepoint_level= new_level; + *sp= 0; + + DBUG_RETURN(0); +} + + +bool federatedx_txn::stmt_begin() +{ + bool result= FALSE; + DBUG_ENTER("federatedx_txn::stmt_begin"); + + if (!savepoint_stmt) + { + if (!savepoint_next) + { + savepoint_next++; + savepoint_level= savepoint_stmt= 0; + } + result= sp_acquire(&savepoint_stmt); + } + + DBUG_RETURN(result); +} + + +int federatedx_txn::stmt_commit() +{ + int result= 0; + DBUG_ENTER("federatedx_txn::stmt_commit"); + + if (savepoint_stmt == 1) + { + savepoint_stmt= 0; + result= txn_commit(); + } + else + if (savepoint_stmt) + result= sp_release(&savepoint_stmt); + + DBUG_RETURN(result); +} + + +int federatedx_txn::stmt_rollback() +{ + int result= 0; + DBUG_ENTER("federated:txn::stmt_rollback"); + + if (savepoint_stmt == 1) + { + savepoint_stmt= 0; + result= txn_rollback(); + } + else + if (savepoint_stmt) + { + result= sp_rollback(&savepoint_stmt); + sp_release(&savepoint_stmt); + } + + DBUG_RETURN(result); +} + + +void federatedx_txn::stmt_autocommit() +{ + federatedx_io *io; + DBUG_ENTER("federatedx_txn::stmt_autocommit"); + + for (io= txn_list; savepoint_stmt && io; io= io->txn_next) + { + if (io->readonly) + continue; + + io->savepoint_restrict(savepoint_stmt); + } + + DBUG_VOID_RETURN; +} + + === added file 'storage/federatedx/ha_federatedx.cc' --- a/storage/federatedx/ha_federatedx.cc 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/ha_federatedx.cc 2009-10-30 18:50:56 +0000 @@ -0,0 +1,3487 @@ +/* +Copyright (c) 2008, Patrick Galbraith +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Neither the name of Patrick Galbraith nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ +/* + + FederatedX Pluggable Storage Engine + + ha_federatedx.cc - FederatedX Pluggable Storage Engine + Patrick Galbraith, 2008 + + This is a handler which uses a foreign database as the data file, as + opposed to a handler like MyISAM, which uses .MYD files locally. + + How this handler works + ---------------------------------- + Normal database files are local and as such: You create a table called + 'users', a file such as 'users.MYD' is created. A handler reads, inserts, + deletes, updates data in this file. The data is stored in particular format, + so to read, that data has to be parsed into fields, to write, fields have to + be stored in this format to write to this data file. + + With FederatedX storage engine, there will be no local files + for each table's data (such as .MYD). A foreign database will store + the data that would normally be in this file. This will necessitate + the use of MySQL client API to read, delete, update, insert this + data. The data will have to be retrieve via an SQL call "SELECT * + FROM users". Then, to read this data, it will have to be retrieved + via mysql_fetch_row one row at a time, then converted from the + column in this select into the format that the handler expects. + + The create table will simply create the .frm file, and within the + "CREATE TABLE" SQL, there SHALL be any of the following : + + connection=scheme://username:password@hostname:port/database/tablename + connection=scheme://username@hostname/database/tablename + connection=scheme://username:password@hostname/database/tablename + connection=scheme://username:password@hostname/database/tablename + + - OR - + + As of 5.1 federatedx now allows you to use a non-url + format, taking advantage of mysql.servers: + + connection="connection_one" + connection="connection_one/table_foo" + + An example would be: + + connection=mysql://username:password@hostname:port/database/tablename + + or, if we had: + + create server 'server_one' foreign data wrapper 'mysql' options + (HOST '127.0.0.1', + DATABASE 'db1', + USER 'root', + PASSWORD '', + PORT 3306, + SOCKET '', + OWNER 'root'); + + CREATE TABLE federatedx.t1 ( + `id` int(20) NOT NULL, + `name` varchar(64) NOT NULL default '' + ) + ENGINE="FEDERATEDX" DEFAULT CHARSET=latin1 + CONNECTION='server_one'; + + So, this will have been the equivalent of + + CONNECTION="mysql://root@127.0.0.1:3306/db1/t1" + + Then, we can also change the server to point to a new schema: + + ALTER SERVER 'server_one' options(DATABASE 'db2'); + + All subsequent calls will now be against db2.t1! Guess what? You don't + have to perform an alter table! + + This connecton="connection string" is necessary for the handler to be + able to connect to the foreign server, either by URL, or by server + name. + + + The basic flow is this: + + SQL calls issues locally -> + mysql handler API (data in handler format) -> + mysql client API (data converted to SQL calls) -> + foreign database -> mysql client API -> + convert result sets (if any) to handler format -> + handler API -> results or rows affected to local + + What this handler does and doesn't support + ------------------------------------------ + * Tables MUST be created on the foreign server prior to any action on those + tables via the handler, first version. IMPORTANT: IF you MUST use the + federatedx storage engine type on the REMOTE end, MAKE SURE [ :) ] That + the table you connect to IS NOT a table pointing BACK to your ORIGNAL + table! You know and have heard the screaching of audio feedback? You + know putting two mirror in front of each other how the reflection + continues for eternity? Well, need I say more?! + * There will not be support for transactions. + * There is no way for the handler to know if the foreign database or table + has changed. The reason for this is that this database has to work like a + data file that would never be written to by anything other than the + database. The integrity of the data in the local table could be breached + if there was any change to the foreign database. + * Support for SELECT, INSERT, UPDATE , DELETE, indexes. + * No ALTER TABLE, DROP TABLE or any other Data Definition Language calls. + * Prepared statements will not be used in the first implementation, it + remains to to be seen whether the limited subset of the client API for the + server supports this. + * This uses SELECT, INSERT, UPDATE, DELETE and not HANDLER for its + implementation. + * This will not work with the query cache. + + Method calls + + A two column table, with one record: + + (SELECT) + + "SELECT * FROM foo" + ha_federatedx::info + ha_federatedx::scan_time: + ha_federatedx::rnd_init: share->select_query SELECT * FROM foo + ha_federatedx::extra + + <for every row of data retrieved> + ha_federatedx::rnd_next + ha_federatedx::convert_row_to_internal_format + ha_federatedx::rnd_next + </for every row of data retrieved> + + ha_federatedx::rnd_end + ha_federatedx::extra + ha_federatedx::reset + + (INSERT) + + "INSERT INTO foo (id, ts) VALUES (2, now());" + + ha_federatedx::write_row + + ha_federatedx::reset + + (UPDATE) + + "UPDATE foo SET ts = now() WHERE id = 1;" + + ha_federatedx::index_init + ha_federatedx::index_read + ha_federatedx::index_read_idx + ha_federatedx::rnd_next + ha_federatedx::convert_row_to_internal_format + ha_federatedx::update_row + + ha_federatedx::extra + ha_federatedx::extra + ha_federatedx::extra + ha_federatedx::external_lock + ha_federatedx::reset + + + How do I use this handler? + -------------------------- + + <insert text about plugin storage engine> + + Next, to use this handler, it's very simple. You must + have two databases running, either both on the same host, or + on different hosts. + + One the server that will be connecting to the foreign + host (client), you create your table as such: + + CREATE TABLE test_table ( + id int(20) NOT NULL auto_increment, + name varchar(32) NOT NULL default '', + other int(20) NOT NULL default '0', + PRIMARY KEY (id), + KEY name (name), + KEY other_key (other)) + ENGINE="FEDERATEDX" + DEFAULT CHARSET=latin1 + CONNECTION='mysql://root@127.0.0.1:9306/federatedx/test_federatedx'; + + Notice the "COMMENT" and "ENGINE" field? This is where you + respectively set the engine type, "FEDERATEDX" and foreign + host information, this being the database your 'client' database + will connect to and use as the "data file". Obviously, the foreign + database is running on port 9306, so you want to start up your other + database so that it is indeed on port 9306, and your federatedx + database on a port other than that. In my setup, I use port 5554 + for federatedx, and port 5555 for the foreign database. + + Then, on the foreign database: + + CREATE TABLE test_table ( + id int(20) NOT NULL auto_increment, + name varchar(32) NOT NULL default '', + other int(20) NOT NULL default '0', + PRIMARY KEY (id), + KEY name (name), + KEY other_key (other)) + ENGINE="<NAME>" <-- whatever you want, or not specify + DEFAULT CHARSET=latin1 ; + + This table is exactly the same (and must be exactly the same), + except that it is not using the federatedx handler and does + not need the URL. + + + How to see the handler in action + -------------------------------- + + When developing this handler, I compiled the federatedx database with + debugging: + + ./configure --with-federatedx-storage-engine + --prefix=/home/mysql/mysql-build/federatedx/ --with-debug + + Once compiled, I did a 'make install' (not for the purpose of installing + the binary, but to install all the files the binary expects to see in the + diretory I specified in the build with --prefix, + "/home/mysql/mysql-build/federatedx". + + Then, I started the foreign server: + + /usr/local/mysql/bin/mysqld_safe + --user=mysql --log=/tmp/mysqld.5555.log -P 5555 + + Then, I went back to the directory containing the newly compiled mysqld, + <builddir>/sql/, started up gdb: + + gdb ./mysqld + + Then, withn the (gdb) prompt: + (gdb) run --gdb --port=5554 --socket=/tmp/mysqld.5554 --skip-innodb --debug + + Next, I open several windows for each: + + 1. Tail the debug trace: tail -f /tmp/mysqld.trace|grep ha_fed + 2. Tail the SQL calls to the foreign database: tail -f /tmp/mysqld.5555.log + 3. A window with a client open to the federatedx server on port 5554 + 4. A window with a client open to the federatedx server on port 5555 + + I would create a table on the client to the foreign server on port + 5555, and then to the federatedx server on port 5554. At this point, + I would run whatever queries I wanted to on the federatedx server, + just always remembering that whatever changes I wanted to make on + the table, or if I created new tables, that I would have to do that + on the foreign server. + + Another thing to look for is 'show variables' to show you that you have + support for federatedx handler support: + + show variables like '%federat%' + + and: + + show storage engines; + + Both should display the federatedx storage handler. + + + Testing + ------- + + Testing for FederatedX as a pluggable storage engine for + now is a manual process that I intend to build a test + suite that works for all pluggable storage engines. + + How to test + + 1. cp fed.dat /tmp + (make sure you have access to "test". Use a user that has + super privileges for now) + 2. mysql -f -u root test < federated.test > federated.myresult 2>&1 + 3. diff federated.result federated.myresult (there _should_ be no differences) + + +*/ + + +#define MYSQL_SERVER 1q +#include "mysql_priv.h" +#include <mysql/plugin.h> + +#ifdef USE_PRAGMA_IMPLEMENTATION +#pragma implementation // gcc: Class implementation +#endif + +#include "ha_federatedx.h" + +#include "m_string.h" + +#include <mysql/plugin.h> + +/* Variables for federatedx share methods */ +static HASH federatedx_open_tables; // To track open tables +static HASH federatedx_open_servers; // To track open servers +pthread_mutex_t federatedx_mutex; // To init the hash +const char ident_quote_char= '`'; // Character for quoting + // identifiers +const char value_quote_char= '\''; // Character for quoting + // literals +static const int bulk_padding= 64; // bytes "overhead" in packet + +/* Variables used when chopping off trailing characters */ +static const uint sizeof_trailing_comma= sizeof(", ") - 1; +static const uint sizeof_trailing_closeparen= sizeof(") ") - 1; +static const uint sizeof_trailing_and= sizeof(" AND ") - 1; +static const uint sizeof_trailing_where= sizeof(" WHERE ") - 1; + +/* Static declaration for handerton */ +static handler *federatedx_create_handler(handlerton *hton, + TABLE_SHARE *table, + MEM_ROOT *mem_root); + +/* FederatedX storage engine handlerton */ + +static handler *federatedx_create_handler(handlerton *hton, + TABLE_SHARE *table, + MEM_ROOT *mem_root) +{ + return new (mem_root) ha_federatedx(hton, table); +} + + +/* Function we use in the creation of our hash to get key */ + +static uchar * +federatedx_share_get_key(FEDERATEDX_SHARE *share, size_t *length, + my_bool not_used __attribute__ ((unused))) +{ + *length= share->share_key_length; + return (uchar*) share->share_key; +} + + +static uchar * +federatedx_server_get_key(FEDERATEDX_SERVER *server, size_t *length, + my_bool not_used __attribute__ ((unused))) +{ + *length= server->key_length; + return server->key; +} + + +/* + Initialize the federatedx handler. + + SYNOPSIS + federatedx_db_init() + p Handlerton + + RETURN + FALSE OK + TRUE Error +*/ + +int federatedx_db_init(void *p) +{ + DBUG_ENTER("federatedx_db_init"); + handlerton *federatedx_hton= (handlerton *)p; + federatedx_hton->state= SHOW_OPTION_YES; + /* This is no longer needed for plugin storage engines */ + federatedx_hton->db_type= DB_TYPE_DEFAULT; + federatedx_hton->savepoint_offset= sizeof(ulong); + federatedx_hton->close_connection= ha_federatedx::disconnect; + federatedx_hton->savepoint_set= ha_federatedx::savepoint_set; + federatedx_hton->savepoint_rollback= ha_federatedx::savepoint_rollback; + federatedx_hton->savepoint_release= ha_federatedx::savepoint_release; + federatedx_hton->commit= ha_federatedx::commit; + federatedx_hton->rollback= ha_federatedx::rollback; + federatedx_hton->create= federatedx_create_handler; + federatedx_hton->flags= HTON_ALTER_NOT_SUPPORTED | HTON_NO_PARTITION; + + if (pthread_mutex_init(&federatedx_mutex, MY_MUTEX_INIT_FAST)) + goto error; + if (!hash_init(&federatedx_open_tables, &my_charset_bin, 32, 0, 0, + (hash_get_key) federatedx_share_get_key, 0, 0) && + !hash_init(&federatedx_open_servers, &my_charset_bin, 32, 0, 0, + (hash_get_key) federatedx_server_get_key, 0, 0)) + { + DBUG_RETURN(FALSE); + } + + VOID(pthread_mutex_destroy(&federatedx_mutex)); +error: + DBUG_RETURN(TRUE); +} + + +/* + Release the federatedx handler. + + SYNOPSIS + federatedx_db_end() + + RETURN + FALSE OK +*/ + +int federatedx_done(void *p) +{ + hash_free(&federatedx_open_tables); + hash_free(&federatedx_open_servers); + VOID(pthread_mutex_destroy(&federatedx_mutex)); + + return 0; +} + +/** + @brief Append identifiers to the string. + + @param[in,out] string The target string. + @param[in] name Identifier name + @param[in] length Length of identifier name in bytes + @param[in] quote_char Quote char to use for quoting identifier. + + @return Operation Status + @retval FALSE OK + @retval TRUE There was an error appending to the string. + + @note This function is based upon the append_identifier() function + in sql_show.cc except that quoting always occurs. +*/ + +bool append_ident(String *string, const char *name, uint length, + const char quote_char) +{ + bool result; + uint clen; + const char *name_end; + DBUG_ENTER("append_ident"); + + if (quote_char) + { + string->reserve(length * 2 + 2); + if ((result= string->append(&quote_char, 1, system_charset_info))) + goto err; + + for (name_end= name+length; name < name_end; name+= clen) + { + uchar c= *(uchar *) name; + if (!(clen= my_mbcharlen(system_charset_info, c))) + clen= 1; + if (clen == 1 && c == (uchar) quote_char && + (result= string->append(&quote_char, 1, system_charset_info))) + goto err; + if ((result= string->append(name, clen, string->charset()))) + goto err; + } + result= string->append(&quote_char, 1, system_charset_info); + } + else + result= string->append(name, length, system_charset_info); + +err: + DBUG_RETURN(result); +} + + +static int parse_url_error(FEDERATEDX_SHARE *share, TABLE *table, int error_num) +{ + char buf[FEDERATEDX_QUERY_BUFFER_SIZE]; + int buf_len; + DBUG_ENTER("ha_federatedx parse_url_error"); + + buf_len= min(table->s->connect_string.length, + FEDERATEDX_QUERY_BUFFER_SIZE-1); + strmake(buf, table->s->connect_string.str, buf_len); + my_error(error_num, MYF(0), buf); + DBUG_RETURN(error_num); +} + +/* + retrieve server object which contains server meta-data + from the system table given a server's name, set share + connection parameter members +*/ +int get_connection(MEM_ROOT *mem_root, FEDERATEDX_SHARE *share) +{ + int error_num= ER_FOREIGN_SERVER_DOESNT_EXIST; + char error_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + FOREIGN_SERVER *server, server_buffer; + DBUG_ENTER("ha_federatedx::get_connection"); + + /* + get_server_by_name() clones the server if exists and allocates + copies of strings in the supplied mem_root + */ + if (!(server= + get_server_by_name(mem_root, share->connection_string, &server_buffer))) + { + DBUG_PRINT("info", ("get_server_by_name returned > 0 error condition!")); + /* need to come up with error handling */ + error_num=1; + goto error; + } + DBUG_PRINT("info", ("get_server_by_name returned server at %lx", + (long unsigned int) server)); + + /* + Most of these should never be empty strings, error handling will + need to be implemented. Also, is this the best way to set the share + members? Is there some allocation needed? In running this code, it works + except there are errors in the trace file of the share being overrun + at the address of the share. + */ + share->server_name_length= server->server_name_length; + share->server_name= server->server_name; + share->username= server->username; + share->password= server->password; + share->database= server->db; +#ifndef I_AM_PARANOID + share->port= server->port > 0 && server->port < 65536 ? +#else + share->port= server->port > 1023 && server->port < 65536 ? +#endif + (ushort) server->port : MYSQL_PORT; + share->hostname= server->host; + if (!(share->socket= server->socket) && + !strcmp(share->hostname, my_localhost)) + share->socket= (char *) MYSQL_UNIX_ADDR; + share->scheme= server->scheme; + + DBUG_PRINT("info", ("share->username: %s", share->username)); + DBUG_PRINT("info", ("share->password: %s", share->password)); + DBUG_PRINT("info", ("share->hostname: %s", share->hostname)); + DBUG_PRINT("info", ("share->database: %s", share->database)); + DBUG_PRINT("info", ("share->port: %d", share->port)); + DBUG_PRINT("info", ("share->socket: %s", share->socket)); + DBUG_RETURN(0); + +error: + my_sprintf(error_buffer, + (error_buffer, "server name: '%s' doesn't exist!", + share->connection_string)); + my_error(error_num, MYF(0), error_buffer); + DBUG_RETURN(error_num); +} + +/* + Parse connection info from table->s->connect_string + + SYNOPSIS + parse_url() + mem_root MEM_ROOT pointer for memory allocation + share pointer to FEDERATEDX share + table pointer to current TABLE class + table_create_flag determines what error to throw + + DESCRIPTION + Populates the share with information about the connection + to the foreign database that will serve as the data source. + This string must be specified (currently) in the "CONNECTION" field, + listed in the CREATE TABLE statement. + + This string MUST be in the format of any of these: + + CONNECTION="scheme://username:password@hostname:port/database/table" + CONNECTION="scheme://username@hostname/database/table" + CONNECTION="scheme://username@hostname:port/database/table" + CONNECTION="scheme://username:password@hostname/database/table" + + _OR_ + + CONNECTION="connection name" + + + + An Example: + + CREATE TABLE t1 (id int(32)) + ENGINE="FEDERATEDX" + CONNECTION="mysql://joe:joespass@192.168.1.111:9308/federatedx/testtable"; + + CREATE TABLE t2 ( + id int(4) NOT NULL auto_increment, + name varchar(32) NOT NULL, + PRIMARY KEY(id) + ) ENGINE="FEDERATEDX" CONNECTION="my_conn"; + + ***IMPORTANT*** + Currently, the FederatedX Storage Engine only supports connecting to another + Database ("scheme" of "mysql"). Connections using JDBC as well as + other connectors are in the planning stage. + + + 'password' and 'port' are both optional. + + RETURN VALUE + 0 success + error_num particular error code + +*/ + +static int parse_url(MEM_ROOT *mem_root, FEDERATEDX_SHARE *share, TABLE *table, + uint table_create_flag) +{ + uint error_num= (table_create_flag ? + ER_FOREIGN_DATA_STRING_INVALID_CANT_CREATE : + ER_FOREIGN_DATA_STRING_INVALID); + DBUG_ENTER("ha_federatedx::parse_url"); + + share->port= 0; + share->socket= 0; + DBUG_PRINT("info", ("share at %lx", (long unsigned int) share)); + DBUG_PRINT("info", ("Length: %u", (uint) table->s->connect_string.length)); + DBUG_PRINT("info", ("String: '%.*s'", (int) table->s->connect_string.length, + table->s->connect_string.str)); + share->connection_string= strmake_root(mem_root, table->s->connect_string.str, + table->s->connect_string.length); + + DBUG_PRINT("info",("parse_url alloced share->connection_string %lx", + (long unsigned int) share->connection_string)); + + DBUG_PRINT("info",("share->connection_string: %s",share->connection_string)); + /* + No :// or @ in connection string. Must be a straight connection name of + either "servername" or "servername/tablename" + */ + if ((!strstr(share->connection_string, "://") && + (!strchr(share->connection_string, '@')))) + { + + DBUG_PRINT("info", + ("share->connection_string: %s internal format " + "share->connection_string: %lx", + share->connection_string, + (ulong) share->connection_string)); + + /* ok, so we do a little parsing, but not completely! */ + share->parsed= FALSE; + /* + If there is a single '/' in the connection string, this means the user is + specifying a table name + */ + + if ((share->table_name= strchr(share->connection_string, '/'))) + { + *share->table_name++= '\0'; + share->table_name_length= strlen(share->table_name); + + DBUG_PRINT("info", + ("internal format, parsed table_name " + "share->connection_string: %s share->table_name: %s", + share->connection_string, share->table_name)); + + /* + there better not be any more '/'s ! + */ + if (strchr(share->table_name, '/')) + goto error; + } + /* + Otherwise, straight server name, use tablename of federatedx table + as remote table name + */ + else + { + /* + Connection specifies everything but, resort to + expecting remote and foreign table names to match + */ + share->table_name= strmake_root(mem_root, table->s->table_name.str, + (share->table_name_length= + table->s->table_name.length)); + DBUG_PRINT("info", + ("internal format, default table_name " + "share->connection_string: %s share->table_name: %s", + share->connection_string, share->table_name)); + } + + if ((error_num= get_connection(mem_root, share))) + goto error; + } + else + { + share->parsed= TRUE; + // Add a null for later termination of table name + share->connection_string[table->s->connect_string.length]= 0; + share->scheme= share->connection_string; + DBUG_PRINT("info",("parse_url alloced share->scheme: %lx", + (ulong) share->scheme)); + + /* + Remove addition of null terminator and store length + for each string in share + */ + if (!(share->username= strstr(share->scheme, "://"))) + goto error; + share->scheme[share->username - share->scheme]= '\0'; + + if (!federatedx_io::handles_scheme(share->scheme)) + goto error; + + share->username+= 3; + + if (!(share->hostname= strchr(share->username, '@'))) + goto error; + *share->hostname++= '\0'; // End username + + if ((share->password= strchr(share->username, ':'))) + { + *share->password++= '\0'; // End username + + /* make sure there isn't an extra / or @ */ + if ((strchr(share->password, '/') || strchr(share->hostname, '@'))) + goto error; + /* + Found that if the string is: + user:@hostname:port/db/table + Then password is a null string, so set to NULL + */ + if ((share->password[0] == '\0')) + share->password= NULL; + } + + /* make sure there isn't an extra / or @ */ + if ((strchr(share->username, '/')) || (strchr(share->hostname, '@'))) + goto error; + + if (!(share->database= strchr(share->hostname, '/'))) + goto error; + *share->database++= '\0'; + + if ((share->sport= strchr(share->hostname, ':'))) + { + *share->sport++= '\0'; + if (share->sport[0] == '\0') + share->sport= NULL; + else + share->port= atoi(share->sport); + } + + if (!(share->table_name= strchr(share->database, '/'))) + goto error; + *share->table_name++= '\0'; + + share->table_name_length= strlen(share->table_name); + + /* make sure there's not an extra / */ + if ((strchr(share->table_name, '/'))) + goto error; + + if (share->hostname[0] == '\0') + share->hostname= NULL; + + } + if (!share->port) + { + if (!share->hostname || strcmp(share->hostname, my_localhost) == 0) + share->socket= (char *) MYSQL_UNIX_ADDR; + else + share->port= MYSQL_PORT; + } + + DBUG_PRINT("info", + ("scheme: %s username: %s password: %s hostname: %s " + "port: %d db: %s tablename: %s", + share->scheme, share->username, share->password, + share->hostname, share->port, share->database, + share->table_name)); + + DBUG_RETURN(0); + +error: + DBUG_RETURN(parse_url_error(share, table, error_num)); +} + +/***************************************************************************** +** FEDERATEDX tables +*****************************************************************************/ + +ha_federatedx::ha_federatedx(handlerton *hton, + TABLE_SHARE *table_arg) + :handler(hton, table_arg), + txn(0), io(0), stored_result(0) +{ + bzero(&bulk_insert, sizeof(bulk_insert)); +} + + +/* + Convert MySQL result set row to handler internal format + + SYNOPSIS + convert_row_to_internal_format() + record Byte pointer to record + row MySQL result set row from fetchrow() + result Result set to use + + DESCRIPTION + This method simply iterates through a row returned via fetchrow with + values from a successful SELECT , and then stores each column's value + in the field object via the field object pointer (pointing to the table's + array of field object pointers). This is how the handler needs the data + to be stored to then return results back to the user + + RETURN VALUE + 0 After fields have had field values stored from record +*/ + +uint ha_federatedx::convert_row_to_internal_format(uchar *record, + FEDERATEDX_IO_ROW *row, + FEDERATEDX_IO_RESULT *result) +{ + ulong *lengths; + Field **field; + int column= 0; + my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set); + DBUG_ENTER("ha_federatedx::convert_row_to_internal_format"); + + lengths= io->fetch_lengths(result); + + for (field= table->field; *field; field++, column++) + { + /* + index variable to move us through the row at the + same iterative step as the field + */ + my_ptrdiff_t old_ptr; + old_ptr= (my_ptrdiff_t) (record - table->record[0]); + (*field)->move_field_offset(old_ptr); + if (io->is_column_null(row, column)) + (*field)->set_null(); + else + { + if (bitmap_is_set(table->read_set, (*field)->field_index)) + { + (*field)->set_notnull(); + (*field)->store(io->get_column_data(row, column), lengths[column], &my_charset_bin); + } + } + (*field)->move_field_offset(-old_ptr); + } + dbug_tmp_restore_column_map(table->write_set, old_map); + DBUG_RETURN(0); +} + +static bool emit_key_part_name(String *to, KEY_PART_INFO *part) +{ + DBUG_ENTER("emit_key_part_name"); + if (append_ident(to, part->field->field_name, + strlen(part->field->field_name), ident_quote_char)) + DBUG_RETURN(1); // Out of memory + DBUG_RETURN(0); +} + +static bool emit_key_part_element(String *to, KEY_PART_INFO *part, + bool needs_quotes, bool is_like, + const uchar *ptr, uint len) +{ + Field *field= part->field; + DBUG_ENTER("emit_key_part_element"); + + if (needs_quotes && to->append(STRING_WITH_LEN("'"))) + DBUG_RETURN(1); + + if (part->type == HA_KEYTYPE_BIT) + { + char buff[STRING_BUFFER_USUAL_SIZE], *buf= buff; + + *buf++= '0'; + *buf++= 'x'; + buf= octet2hex(buf, (char*) ptr, len); + if (to->append((char*) buff, (uint)(buf - buff))) + DBUG_RETURN(1); + } + else if (part->key_part_flag & HA_BLOB_PART) + { + String blob; + uint blob_length= uint2korr(ptr); + blob.set_quick((char*) ptr+HA_KEY_BLOB_LENGTH, + blob_length, &my_charset_bin); + if (append_escaped(to, &blob)) + DBUG_RETURN(1); + } + else if (part->key_part_flag & HA_VAR_LENGTH_PART) + { + String varchar; + uint var_length= uint2korr(ptr); + varchar.set_quick((char*) ptr+HA_KEY_BLOB_LENGTH, + var_length, &my_charset_bin); + if (append_escaped(to, &varchar)) + DBUG_RETURN(1); + } + else + { + char strbuff[MAX_FIELD_WIDTH]; + String str(strbuff, sizeof(strbuff), part->field->charset()), *res; + + res= field->val_str(&str, ptr); + + if (field->result_type() == STRING_RESULT) + { + if (append_escaped(to, res)) + DBUG_RETURN(1); + } + else if (to->append(res->ptr(), res->length())) + DBUG_RETURN(1); + } + + if (is_like && to->append(STRING_WITH_LEN("%"))) + DBUG_RETURN(1); + + if (needs_quotes && to->append(STRING_WITH_LEN("'"))) + DBUG_RETURN(1); + + DBUG_RETURN(0); +} + +/* + Create a WHERE clause based off of values in keys + Note: This code was inspired by key_copy from key.cc + + SYNOPSIS + create_where_from_key () + to String object to store WHERE clause + key_info KEY struct pointer + key byte pointer containing key + key_length length of key + range_type 0 - no range, 1 - min range, 2 - max range + (see enum range_operation) + + DESCRIPTION + Using iteration through all the keys via a KEY_PART_INFO pointer, + This method 'extracts' the value of each key in the byte pointer + *key, and for each key found, constructs an appropriate WHERE clause + + RETURN VALUE + 0 After all keys have been accounted for to create the WHERE clause + 1 No keys found + + Range flags Table per Timour: + + ----------------- + - start_key: + * ">" -> HA_READ_AFTER_KEY + * ">=" -> HA_READ_KEY_OR_NEXT + * "=" -> HA_READ_KEY_EXACT + + - end_key: + * "<" -> HA_READ_BEFORE_KEY + * "<=" -> HA_READ_AFTER_KEY + + records_in_range: + ----------------- + - start_key: + * ">" -> HA_READ_AFTER_KEY + * ">=" -> HA_READ_KEY_EXACT + * "=" -> HA_READ_KEY_EXACT + + - end_key: + * "<" -> HA_READ_BEFORE_KEY + * "<=" -> HA_READ_AFTER_KEY + * "=" -> HA_READ_AFTER_KEY + +0 HA_READ_KEY_EXACT, Find first record else error +1 HA_READ_KEY_OR_NEXT, Record or next record +2 HA_READ_KEY_OR_PREV, Record or previous +3 HA_READ_AFTER_KEY, Find next rec. after key-record +4 HA_READ_BEFORE_KEY, Find next rec. before key-record +5 HA_READ_PREFIX, Key which as same prefix +6 HA_READ_PREFIX_LAST, Last key with the same prefix +7 HA_READ_PREFIX_LAST_OR_PREV, Last or prev key with the same prefix + +Flags that I've found: + +id, primary key, varchar + +id = 'ccccc' +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 0 end_key NULL + +id > 'ccccc' +records_in_range: start_key 3 end_key NULL +read_range_first: start_key 3 end_key NULL + +id < 'ccccc' +records_in_range: start_key NULL end_key 4 +read_range_first: start_key NULL end_key 4 + +id <= 'ccccc' +records_in_range: start_key NULL end_key 3 +read_range_first: start_key NULL end_key 3 + +id >= 'ccccc' +records_in_range: start_key 0 end_key NULL +read_range_first: start_key 1 end_key NULL + +id like 'cc%cc' +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 1 end_key 3 + +id > 'aaaaa' and id < 'ccccc' +records_in_range: start_key 3 end_key 4 +read_range_first: start_key 3 end_key 4 + +id >= 'aaaaa' and id < 'ccccc'; +records_in_range: start_key 0 end_key 4 +read_range_first: start_key 1 end_key 4 + +id >= 'aaaaa' and id <= 'ccccc'; +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 1 end_key 3 + +id > 'aaaaa' and id <= 'ccccc'; +records_in_range: start_key 3 end_key 3 +read_range_first: start_key 3 end_key 3 + +numeric keys: + +id = 4 +index_read_idx: start_key 0 end_key NULL + +id > 4 +records_in_range: start_key 3 end_key NULL +read_range_first: start_key 3 end_key NULL + +id >= 4 +records_in_range: start_key 0 end_key NULL +read_range_first: start_key 1 end_key NULL + +id < 4 +records_in_range: start_key NULL end_key 4 +read_range_first: start_key NULL end_key 4 + +id <= 4 +records_in_range: start_key NULL end_key 3 +read_range_first: start_key NULL end_key 3 + +id like 4 +full table scan, select * from + +id > 2 and id < 8 +records_in_range: start_key 3 end_key 4 +read_range_first: start_key 3 end_key 4 + +id >= 2 and id < 8 +records_in_range: start_key 0 end_key 4 +read_range_first: start_key 1 end_key 4 + +id >= 2 and id <= 8 +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 1 end_key 3 + +id > 2 and id <= 8 +records_in_range: start_key 3 end_key 3 +read_range_first: start_key 3 end_key 3 + +multi keys (id int, name varchar, other varchar) + +id = 1; +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 0 end_key NULL + +id > 4; +id > 2 and name = '333'; remote: id > 2 +id > 2 and name > '333'; remote: id > 2 +id > 2 and name > '333' and other < 'ddd'; remote: id > 2 no results +id > 2 and name >= '333' and other < 'ddd'; remote: id > 2 1 result +id >= 4 and name = 'eric was here' and other > 'eeee'; +records_in_range: start_key 3 end_key NULL +read_range_first: start_key 3 end_key NULL + +id >= 4; +id >= 2 and name = '333' and other < 'ddd'; +remote: `id` >= 2 AND `name` >= '333'; +records_in_range: start_key 0 end_key NULL +read_range_first: start_key 1 end_key NULL + +id < 4; +id < 3 and name = '222' and other <= 'ccc'; remote: id < 3 +records_in_range: start_key NULL end_key 4 +read_range_first: start_key NULL end_key 4 + +id <= 4; +records_in_range: start_key NULL end_key 3 +read_range_first: start_key NULL end_key 3 + +id like 4; +full table scan + +id > 2 and id < 4; +records_in_range: start_key 3 end_key 4 +read_range_first: start_key 3 end_key 4 + +id >= 2 and id < 4; +records_in_range: start_key 0 end_key 4 +read_range_first: start_key 1 end_key 4 + +id >= 2 and id <= 4; +records_in_range: start_key 0 end_key 3 +read_range_first: start_key 1 end_key 3 + +id > 2 and id <= 4; +id = 6 and name = 'eric was here' and other > 'eeee'; +remote: (`id` > 6 AND `name` > 'eric was here' AND `other` > 'eeee') +AND (`id` <= 6) AND ( AND `name` <= 'eric was here') +no results +records_in_range: start_key 3 end_key 3 +read_range_first: start_key 3 end_key 3 + +Summary: + +* If the start key flag is 0 the max key flag shouldn't even be set, + and if it is, the query produced would be invalid. +* Multipart keys, even if containing some or all numeric columns, + are treated the same as non-numeric keys + + If the query is " = " (quotes or not): + - records in range start key flag HA_READ_KEY_EXACT, + end key flag HA_READ_AFTER_KEY (incorrect) + - any other: start key flag HA_READ_KEY_OR_NEXT, + end key flag HA_READ_AFTER_KEY (correct) + +* 'like' queries (of key) + - Numeric, full table scan + - Non-numeric + records_in_range: start_key 0 end_key 3 + other : start_key 1 end_key 3 + +* If the key flag is HA_READ_AFTER_KEY: + if start_key, append > + if end_key, append <= + +* If create_where_key was called by records_in_range: + + - if the key is numeric: + start key flag is 0 when end key is NULL, end key flag is 3 or 4 + - if create_where_key was called by any other function: + start key flag is 1 when end key is NULL, end key flag is 3 or 4 + - if the key is non-numeric, or multipart + When the query is an exact match, the start key flag is 0, + end key flag is 3 for what should be a no-range condition where + you should have 0 and max key NULL, which it is if called by + read_range_first + +Conclusion: + +1. Need logic to determin if a key is min or max when the flag is +HA_READ_AFTER_KEY, and handle appending correct operator accordingly + +2. Need a boolean flag to pass to create_where_from_key, used in the +switch statement. Add 1 to the flag if: + - start key flag is HA_READ_KEY_EXACT and the end key is NULL + +*/ + +bool ha_federatedx::create_where_from_key(String *to, + KEY *key_info, + const key_range *start_key, + const key_range *end_key, + bool from_records_in_range, + bool eq_range) +{ + bool both_not_null= + (start_key != NULL && end_key != NULL) ? TRUE : FALSE; + const uchar *ptr; + uint remainder, length; + char tmpbuff[FEDERATEDX_QUERY_BUFFER_SIZE]; + String tmp(tmpbuff, sizeof(tmpbuff), system_charset_info); + const key_range *ranges[2]= { start_key, end_key }; + my_bitmap_map *old_map; + DBUG_ENTER("ha_federatedx::create_where_from_key"); + + tmp.length(0); + if (start_key == NULL && end_key == NULL) + DBUG_RETURN(1); + + old_map= dbug_tmp_use_all_columns(table, table->write_set); + for (uint i= 0; i <= 1; i++) + { + bool needs_quotes; + KEY_PART_INFO *key_part; + if (ranges[i] == NULL) + continue; + + if (both_not_null) + { + if (i > 0) + tmp.append(STRING_WITH_LEN(") AND (")); + else + tmp.append(STRING_WITH_LEN(" (")); + } + + for (key_part= key_info->key_part, + remainder= key_info->key_parts, + length= ranges[i]->length, + ptr= ranges[i]->key; ; + remainder--, + key_part++) + { + Field *field= key_part->field; + uint store_length= key_part->store_length; + uint part_length= min(store_length, length); + needs_quotes= field->str_needs_quotes(); + DBUG_DUMP("key, start of loop", ptr, length); + + if (key_part->null_bit) + { + if (*ptr++) + { + /* + We got "IS [NOT] NULL" condition against nullable column. We + distinguish between "IS NOT NULL" and "IS NULL" by flag. For + "IS NULL", flag is set to HA_READ_KEY_EXACT. + */ + if (emit_key_part_name(&tmp, key_part) || + tmp.append(ranges[i]->flag == HA_READ_KEY_EXACT ? + " IS NULL " : " IS NOT NULL ")) + goto err; + /* + We need to adjust pointer and length to be prepared for next + key part. As well as check if this was last key part. + */ + goto prepare_for_next_key_part; + } + } + + if (tmp.append(STRING_WITH_LEN(" ("))) + goto err; + + switch (ranges[i]->flag) { + case HA_READ_KEY_EXACT: + DBUG_PRINT("info", ("federatedx HA_READ_KEY_EXACT %d", i)); + if (store_length >= length || + !needs_quotes || + key_part->type == HA_KEYTYPE_BIT || + field->result_type() != STRING_RESULT) + { + if (emit_key_part_name(&tmp, key_part)) + goto err; + + if (from_records_in_range) + { + if (tmp.append(STRING_WITH_LEN(" >= "))) + goto err; + } + else + { + if (tmp.append(STRING_WITH_LEN(" = "))) + goto err; + } + + if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr, + part_length)) + goto err; + } + else + { + /* LIKE */ + if (emit_key_part_name(&tmp, key_part) || + tmp.append(STRING_WITH_LEN(" LIKE ")) || + emit_key_part_element(&tmp, key_part, needs_quotes, 1, ptr, + part_length)) + goto err; + } + break; + case HA_READ_AFTER_KEY: + if (eq_range) + { + if (tmp.append("1=1")) // Dummy + goto err; + break; + } + DBUG_PRINT("info", ("federatedx HA_READ_AFTER_KEY %d", i)); + if (store_length >= length) /* end key */ + { + if (emit_key_part_name(&tmp, key_part)) + goto err; + + if (i > 0) /* end key */ + { + if (tmp.append(STRING_WITH_LEN(" <= "))) + goto err; + } + else /* start key */ + { + if (tmp.append(STRING_WITH_LEN(" > "))) + goto err; + } + + if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr, + part_length)) + { + goto err; + } + break; + } + case HA_READ_KEY_OR_NEXT: + DBUG_PRINT("info", ("federatedx HA_READ_KEY_OR_NEXT %d", i)); + if (emit_key_part_name(&tmp, key_part) || + tmp.append(STRING_WITH_LEN(" >= ")) || + emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr, + part_length)) + goto err; + break; + case HA_READ_BEFORE_KEY: + DBUG_PRINT("info", ("federatedx HA_READ_BEFORE_KEY %d", i)); + if (store_length >= length) + { + if (emit_key_part_name(&tmp, key_part) || + tmp.append(STRING_WITH_LEN(" < ")) || + emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr, + part_length)) + goto err; + break; + } + case HA_READ_KEY_OR_PREV: + DBUG_PRINT("info", ("federatedx HA_READ_KEY_OR_PREV %d", i)); + if (emit_key_part_name(&tmp, key_part) || + tmp.append(STRING_WITH_LEN(" <= ")) || + emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr, + part_length)) + goto err; + break; + default: + DBUG_PRINT("info",("cannot handle flag %d", ranges[i]->flag)); + goto err; + } + if (tmp.append(STRING_WITH_LEN(") "))) + goto err; + +prepare_for_next_key_part: + if (store_length >= length) + break; + DBUG_PRINT("info", ("remainder %d", remainder)); + DBUG_ASSERT(remainder > 1); + length-= store_length; + /* + For nullable columns, null-byte is already skipped before, that is + ptr was incremented by 1. Since store_length still counts null-byte, + we need to subtract 1 from store_length. + */ + ptr+= store_length - test(key_part->null_bit); + if (tmp.append(STRING_WITH_LEN(" AND "))) + goto err; + + DBUG_PRINT("info", + ("create_where_from_key WHERE clause: %s", + tmp.c_ptr_quick())); + } + } + dbug_tmp_restore_column_map(table->write_set, old_map); + + if (both_not_null) + if (tmp.append(STRING_WITH_LEN(") "))) + DBUG_RETURN(1); + + if (to->append(STRING_WITH_LEN(" WHERE "))) + DBUG_RETURN(1); + + if (to->append(tmp)) + DBUG_RETURN(1); + + DBUG_RETURN(0); + +err: + dbug_tmp_restore_column_map(table->write_set, old_map); + DBUG_RETURN(1); +} + +static void fill_server(MEM_ROOT *mem_root, FEDERATEDX_SERVER *server, + FEDERATEDX_SHARE *share, CHARSET_INFO *table_charset) +{ + char buffer[STRING_BUFFER_USUAL_SIZE]; + String key(buffer, sizeof(buffer), &my_charset_bin); + String scheme(share->scheme, &my_charset_latin1); + String hostname(share->hostname, &my_charset_latin1); + String database(share->database, system_charset_info); + String username(share->username, system_charset_info); + String socket(share->socket ? share->socket : "", files_charset_info); + String password(share->password ? share->password : "", &my_charset_bin); + DBUG_ENTER("fill_server"); + + /* Do some case conversions */ + scheme.reserve(scheme.length()); + scheme.length(my_casedn_str(&my_charset_latin1, scheme.c_ptr_safe())); + + hostname.reserve(hostname.length()); + hostname.length(my_casedn_str(&my_charset_latin1, hostname.c_ptr_safe())); + + if (lower_case_table_names) + { + database.reserve(database.length()); + database.length(my_casedn_str(system_charset_info, database.c_ptr_safe())); + } + + if (lower_case_file_system && socket.length()) + { + socket.reserve(socket.length()); + socket.length(my_casedn_str(files_charset_info, socket.c_ptr_safe())); + } + + /* start with all bytes zeroed */ + bzero(server, sizeof(*server)); + + key.length(0); + key.reserve(scheme.length() + hostname.length() + database.length() + + socket.length() + username.length() + password.length() + + sizeof(int) + 8); + key.append(scheme); + key.q_append('\0'); + server->hostname= (const char *) (intptr) key.length(); + key.append(hostname); + key.q_append('\0'); + server->database= (const char *) (intptr) key.length(); + key.append(database); + key.q_append('\0'); + key.q_append((uint32) share->port); + server->socket= (const char *) (intptr) key.length(); + key.append(socket); + key.q_append('\0'); + server->username= (const char *) (intptr) key.length(); + key.append(username); + key.q_append('\0'); + server->password= (const char *) (intptr) key.length(); + key.append(password); + + server->key_length= key.length(); + server->key= (uchar *) memdup_root(mem_root, key.ptr(), key.length()+1); + + /* pointer magic */ + server->scheme+= (intptr) server->key; + server->hostname+= (intptr) server->key; + server->database+= (intptr) server->key; + server->username+= (intptr) server->key; + server->password+= (intptr) server->key; + server->socket+= (intptr) server->key; + server->port= share->port; + + if (!share->socket) + server->socket= NULL; + if (!share->password) + server->password= NULL; + + if (table_charset) + server->csname= strdup_root(mem_root, table_charset->csname); + + DBUG_VOID_RETURN; +} + + +static FEDERATEDX_SERVER *get_server(FEDERATEDX_SHARE *share, TABLE *table) +{ + FEDERATEDX_SERVER *server= NULL, tmp_server; + MEM_ROOT mem_root; + char buffer[STRING_BUFFER_USUAL_SIZE]; + String key(buffer, sizeof(buffer), &my_charset_bin); + String scheme(share->scheme, &my_charset_latin1); + String hostname(share->hostname, &my_charset_latin1); + String database(share->database, system_charset_info); + String username(share->username, system_charset_info); + String socket(share->socket ? share->socket : "", files_charset_info); + String password(share->password ? share->password : "", &my_charset_bin); + DBUG_ENTER("ha_federated.cc::get_server"); + + safe_mutex_assert_owner(&federatedx_mutex); + + init_alloc_root(&mem_root, 4096, 4096); + + fill_server(&mem_root, &tmp_server, share, table ? table->s->table_charset : 0); + + if (!(server= (FEDERATEDX_SERVER *) hash_search(&federatedx_open_servers, + tmp_server.key, + tmp_server.key_length))) + { + if (!table || !tmp_server.csname) + goto error; + + if (!(server= (FEDERATEDX_SERVER *) memdup_root(&mem_root, + (char *) &tmp_server, + sizeof(*server)))) + goto error; + + server->mem_root= mem_root; + + if (my_hash_insert(&federatedx_open_servers, (uchar*) server)) + goto error; + + pthread_mutex_init(&server->mutex, MY_MUTEX_INIT_FAST); + } + else + free_root(&mem_root, MYF(0)); /* prevents memory leak */ + + server->use_count++; + + DBUG_RETURN(server); +error: + free_root(&mem_root, MYF(0)); + DBUG_RETURN(NULL); +} + + +/* + Example of simple lock controls. The "share" it creates is structure we will + pass to each federatedx handler. Do you have to have one of these? Well, you + have pieces that are used for locking, and they are needed to function. +*/ + +static FEDERATEDX_SHARE *get_share(const char *table_name, TABLE *table) +{ + char query_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + Field **field; + String query(query_buffer, sizeof(query_buffer), &my_charset_bin); + FEDERATEDX_SHARE *share= NULL, tmp_share; + MEM_ROOT mem_root; + DBUG_ENTER("ha_federatedx.cc::get_share"); + + /* + In order to use this string, we must first zero it's length, + or it will contain garbage + */ + query.length(0); + + bzero(&tmp_share, sizeof(tmp_share)); + init_alloc_root(&mem_root, 256, 0); + + pthread_mutex_lock(&federatedx_mutex); + + tmp_share.share_key= table_name; + tmp_share.share_key_length= strlen(table_name); + if (parse_url(&mem_root, &tmp_share, table, 0)) + goto error; + + /* TODO: change tmp_share.scheme to LEX_STRING object */ + if (!(share= (FEDERATEDX_SHARE *) hash_search(&federatedx_open_tables, + (uchar*) tmp_share.share_key, + tmp_share. + share_key_length))) + { + query.set_charset(system_charset_info); + query.append(STRING_WITH_LEN("SELECT ")); + for (field= table->field; *field; field++) + { + append_ident(&query, (*field)->field_name, + strlen((*field)->field_name), ident_quote_char); + query.append(STRING_WITH_LEN(", ")); + } + /* chops off trailing comma */ + query.length(query.length() - sizeof_trailing_comma); + + query.append(STRING_WITH_LEN(" FROM ")); + + append_ident(&query, tmp_share.table_name, + tmp_share.table_name_length, ident_quote_char); + + if (!(share= (FEDERATEDX_SHARE *) memdup_root(&mem_root, (char*)&tmp_share, sizeof(*share))) || + !(share->select_query= (char*) strmake_root(&mem_root, query.ptr(), query.length() + 1))) + goto error; + + share->mem_root= mem_root; + + DBUG_PRINT("info", + ("share->select_query %s", share->select_query)); + + if (!(share->s= get_server(share, table))) + goto error; + + if (my_hash_insert(&federatedx_open_tables, (uchar*) share)) + goto error; + thr_lock_init(&share->lock); + } + else + free_root(&mem_root, MYF(0)); /* prevents memory leak */ + + share->use_count++; + pthread_mutex_unlock(&federatedx_mutex); + + DBUG_RETURN(share); + +error: + pthread_mutex_unlock(&federatedx_mutex); + free_root(&mem_root, MYF(0)); + DBUG_RETURN(NULL); +} + + +static int free_server(federatedx_txn *txn, FEDERATEDX_SERVER *server) +{ + bool destroy; + DBUG_ENTER("free_server"); + + pthread_mutex_lock(&federatedx_mutex); + if ((destroy= !--server->use_count)) + hash_delete(&federatedx_open_servers, (uchar*) server); + pthread_mutex_unlock(&federatedx_mutex); + + if (destroy) + { + MEM_ROOT mem_root; + + txn->close(server); + + DBUG_ASSERT(server->io_count == 0); + + pthread_mutex_destroy(&server->mutex); + mem_root= server->mem_root; + free_root(&mem_root, MYF(0)); + } + + DBUG_RETURN(0); +} + + +/* + Free lock controls. We call this whenever we close a table. + If the table had the last reference to the share then we + free memory associated with it. +*/ + +static int free_share(federatedx_txn *txn, FEDERATEDX_SHARE *share) +{ + bool destroy; + DBUG_ENTER("free_share"); + + pthread_mutex_lock(&federatedx_mutex); + if ((destroy= !--share->use_count)) + hash_delete(&federatedx_open_tables, (uchar*) share); + pthread_mutex_unlock(&federatedx_mutex); + + if (destroy) + { + MEM_ROOT mem_root; + FEDERATEDX_SERVER *server= share->s; + + thr_lock_delete(&share->lock); + + mem_root= share->mem_root; + free_root(&mem_root, MYF(0)); + + free_server(txn, server); + } + + DBUG_RETURN(0); +} + + +ha_rows ha_federatedx::records_in_range(uint inx, key_range *start_key, + key_range *end_key) +{ + /* + + We really want indexes to be used as often as possible, therefore + we just need to hard-code the return value to a very low number to + force the issue + +*/ + DBUG_ENTER("ha_federatedx::records_in_range"); + DBUG_RETURN(FEDERATEDX_RECORDS_IN_RANGE); +} +/* + If frm_error() is called then we will use this to to find out + what file extentions exist for the storage engine. This is + also used by the default rename_table and delete_table method + in handler.cc. +*/ + +const char **ha_federatedx::bas_ext() const +{ + static const char *ext[]= + { + NullS + }; + return ext; +} + + +federatedx_txn *ha_federatedx::get_txn(THD *thd, bool no_create) +{ + federatedx_txn **txnp= (federatedx_txn **) ha_data(thd); + if (!*txnp && !no_create) + *txnp= new federatedx_txn(); + return *txnp; +} + + +int ha_federatedx::disconnect(handlerton *hton, MYSQL_THD thd) +{ + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + delete txn; + return 0; +} + + +/* + Used for opening tables. The name will be the name of the file. + A table is opened when it needs to be opened. For instance + when a request comes in for a select on the table (tables are not + open and closed for each request, they are cached). + + Called from handler.cc by handler::ha_open(). The server opens + all tables by calling ha_open() which then calls the handler + specific open(). +*/ + +int ha_federatedx::open(const char *name, int mode, uint test_if_locked) +{ + int error; + THD *thd= current_thd; + DBUG_ENTER("ha_federatedx::open"); + + if (!(share= get_share(name, table))) + DBUG_RETURN(1); + thr_lock_data_init(&share->lock, &lock, NULL); + + DBUG_ASSERT(io == NULL); + + txn= get_txn(thd); + + if ((error= txn->acquire(share, TRUE, &io))) + { + free_share(txn, share); + DBUG_RETURN(error); + } + + txn->release(&io); + + ref_length= (table->s->primary_key != MAX_KEY ? + table->key_info[table->s->primary_key].key_length : + table->s->reclength); + DBUG_PRINT("info", ("ref_length: %u", ref_length)); + + reset(); + + DBUG_RETURN(0); +} + + +/* + Closes a table. We call the free_share() function to free any resources + that we have allocated in the "shared" structure. + + Called from sql_base.cc, sql_select.cc, and table.cc. + In sql_select.cc it is only used to close up temporary tables or during + the process where a temporary table is converted over to being a + myisam table. + For sql_base.cc look at close_data_tables(). +*/ + +int ha_federatedx::close(void) +{ + int retval, error; + THD *thd= current_thd; + DBUG_ENTER("ha_federatedx::close"); + + /* free the result set */ + if (stored_result) + retval= free_result(); + + /* Disconnect from mysql */ + if ((txn= get_txn(thd, true))) + txn->release(&io); + + DBUG_ASSERT(io == NULL); + + if ((error= free_share(txn, share))) + retval= error; + DBUG_RETURN(retval); +} + +/* + + Checks if a field in a record is SQL NULL. + + SYNOPSIS + field_in_record_is_null() + table TABLE pointer, MySQL table object + field Field pointer, MySQL field object + record char pointer, contains record + + DESCRIPTION + This method uses the record format information in table to track + the null bit in record. + + RETURN VALUE + 1 if NULL + 0 otherwise +*/ + +static inline uint field_in_record_is_null(TABLE *table, + Field *field, + char *record) +{ + int null_offset; + DBUG_ENTER("ha_federatedx::field_in_record_is_null"); + + if (!field->null_ptr) + DBUG_RETURN(0); + + null_offset= (uint) ((char*)field->null_ptr - (char*)table->record[0]); + + if (record[null_offset] & field->null_bit) + DBUG_RETURN(1); + + DBUG_RETURN(0); +} + + +/** + @brief Construct the INSERT statement. + + @details This method will construct the INSERT statement and appends it to + the supplied query string buffer. + + @return + @retval FALSE No error + @retval TRUE Failure +*/ + +bool ha_federatedx::append_stmt_insert(String *query) +{ + char insert_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + Field **field; + uint tmp_length; + bool added_field= FALSE; + + /* The main insert query string */ + String insert_string(insert_buffer, sizeof(insert_buffer), &my_charset_bin); + DBUG_ENTER("ha_federatedx::append_stmt_insert"); + + insert_string.length(0); + + if (replace_duplicates) + insert_string.append(STRING_WITH_LEN("REPLACE INTO ")); + else if (ignore_duplicates && !insert_dup_update) + insert_string.append(STRING_WITH_LEN("INSERT IGNORE INTO ")); + else + insert_string.append(STRING_WITH_LEN("INSERT INTO ")); + append_ident(&insert_string, share->table_name, share->table_name_length, + ident_quote_char); + tmp_length= insert_string.length(); + insert_string.append(STRING_WITH_LEN(" (")); + + /* + loop through the field pointer array, add any fields to both the values + list and the fields list that match the current query id + */ + for (field= table->field; *field; field++) + { + if (bitmap_is_set(table->write_set, (*field)->field_index)) + { + /* append the field name */ + append_ident(&insert_string, (*field)->field_name, + strlen((*field)->field_name), ident_quote_char); + + /* append commas between both fields and fieldnames */ + /* + unfortunately, we can't use the logic if *(fields + 1) to + make the following appends conditional as we don't know if the + next field is in the write set + */ + insert_string.append(STRING_WITH_LEN(", ")); + added_field= TRUE; + } + } + + if (added_field) + { + /* Remove trailing comma. */ + insert_string.length(insert_string.length() - sizeof_trailing_comma); + insert_string.append(STRING_WITH_LEN(") ")); + } + else + { + /* If there were no fields, we don't want to add a closing paren. */ + insert_string.length(tmp_length); + } + + insert_string.append(STRING_WITH_LEN(" VALUES ")); + + DBUG_RETURN(query->append(insert_string)); +} + + +/* + write_row() inserts a row. No extra() hint is given currently if a bulk load + is happeneding. buf() is a byte array of data. You can use the field + information to extract the data from the native byte array type. + Example of this would be: + for (Field **field=table->field ; *field ; field++) + { + ... + } + + Called from item_sum.cc, item_sum.cc, sql_acl.cc, sql_insert.cc, + sql_insert.cc, sql_select.cc, sql_table.cc, sql_udf.cc, and sql_update.cc. +*/ + +int ha_federatedx::write_row(uchar *buf) +{ + char values_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + char insert_field_value_buffer[STRING_BUFFER_USUAL_SIZE]; + Field **field; + uint tmp_length; + int error= 0; + bool use_bulk_insert; + bool auto_increment_update_required= (table->next_number_field != NULL); + + /* The string containing the values to be added to the insert */ + String values_string(values_buffer, sizeof(values_buffer), &my_charset_bin); + /* The actual value of the field, to be added to the values_string */ + String insert_field_value_string(insert_field_value_buffer, + sizeof(insert_field_value_buffer), + &my_charset_bin); + my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set); + DBUG_ENTER("ha_federatedx::write_row"); + + values_string.length(0); + insert_field_value_string.length(0); + ha_statistic_increment(&SSV::ha_write_count); + if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT) + table->timestamp_field->set_time(); + + /* + start both our field and field values strings + We must disable multi-row insert for "INSERT...ON DUPLICATE KEY UPDATE" + Ignore duplicates is always true when insert_dup_update is true. + When replace_duplicates == TRUE, we can safely enable multi-row insert. + When performing multi-row insert, we only collect the columns values for + the row. The start of the statement is only created when the first + row is copied in to the bulk_insert string. + */ + if (!(use_bulk_insert= bulk_insert.str && + (!insert_dup_update || replace_duplicates))) + append_stmt_insert(&values_string); + + values_string.append(STRING_WITH_LEN(" (")); + tmp_length= values_string.length(); + + /* + loop through the field pointer array, add any fields to both the values + list and the fields list that is part of the write set + */ + for (field= table->field; *field; field++) + { + if (bitmap_is_set(table->write_set, (*field)->field_index)) + { + if ((*field)->is_null()) + values_string.append(STRING_WITH_LEN(" NULL ")); + else + { + bool needs_quote= (*field)->str_needs_quotes(); + (*field)->val_str(&insert_field_value_string); + if (needs_quote) + values_string.append(value_quote_char); + insert_field_value_string.print(&values_string); + if (needs_quote) + values_string.append(value_quote_char); + + insert_field_value_string.length(0); + } + + /* append commas between both fields and fieldnames */ + /* + unfortunately, we can't use the logic if *(fields + 1) to + make the following appends conditional as we don't know if the + next field is in the write set + */ + values_string.append(STRING_WITH_LEN(", ")); + } + } + dbug_tmp_restore_column_map(table->read_set, old_map); + + /* + if there were no fields, we don't want to add a closing paren + AND, we don't want to chop off the last char '(' + insert will be "INSERT INTO t1 VALUES ();" + */ + if (values_string.length() > tmp_length) + { + /* chops off trailing comma */ + values_string.length(values_string.length() - sizeof_trailing_comma); + } + /* we always want to append this, even if there aren't any fields */ + values_string.append(STRING_WITH_LEN(") ")); + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (use_bulk_insert) + { + /* + Send the current bulk insert out if appending the current row would + cause the statement to overflow the packet size, otherwise set + auto_increment_update_required to FALSE as no query was executed. + */ + if (bulk_insert.length + values_string.length() + bulk_padding > + io->max_query_size() && bulk_insert.length) + { + error= io->query(bulk_insert.str, bulk_insert.length); + bulk_insert.length= 0; + } + else + auto_increment_update_required= FALSE; + + if (bulk_insert.length == 0) + { + char insert_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + String insert_string(insert_buffer, sizeof(insert_buffer), + &my_charset_bin); + insert_string.length(0); + append_stmt_insert(&insert_string); + dynstr_append_mem(&bulk_insert, insert_string.ptr(), + insert_string.length()); + } + else + dynstr_append_mem(&bulk_insert, ",", 1); + + dynstr_append_mem(&bulk_insert, values_string.ptr(), + values_string.length()); + } + else + { + error= io->query(values_string.ptr(), values_string.length()); + } + + if (error) + { + DBUG_RETURN(stash_remote_error()); + } + /* + If the table we've just written a record to contains an auto_increment + field, then store the last_insert_id() value from the foreign server + */ + if (auto_increment_update_required) + { + update_auto_increment(); + + /* mysql_insert() uses this for protocol return value */ + table->next_number_field->store(stats.auto_increment_value, 1); + } + + DBUG_RETURN(0); +} + + +/** + @brief Prepares the storage engine for bulk inserts. + + @param[in] rows estimated number of rows in bulk insert + or 0 if unknown. + + @details Initializes memory structures required for bulk insert. +*/ + +void ha_federatedx::start_bulk_insert(ha_rows rows) +{ + uint page_size; + DBUG_ENTER("ha_federatedx::start_bulk_insert"); + + dynstr_free(&bulk_insert); + + /** + We don't bother with bulk-insert semantics when the estimated rows == 1 + The rows value will be 0 if the server does not know how many rows + would be inserted. This can occur when performing INSERT...SELECT + */ + + if (rows == 1) + DBUG_VOID_RETURN; + + /* + Make sure we have an open connection so that we know the + maximum packet size. + */ + if (txn->acquire(share, FALSE, &io)) + DBUG_VOID_RETURN; + + page_size= (uint) my_getpagesize(); + + if (init_dynamic_string(&bulk_insert, NULL, page_size, page_size)) + DBUG_VOID_RETURN; + + bulk_insert.length= 0; + DBUG_VOID_RETURN; +} + + +/** + @brief End bulk insert. + + @details This method will send any remaining rows to the remote server. + Finally, it will deinitialize the bulk insert data structure. + + @return Operation status + @retval 0 No error + @retval != 0 Error occured at remote server. Also sets my_errno. +*/ + +int ha_federatedx::end_bulk_insert(bool abort) +{ + int error= 0; + DBUG_ENTER("ha_federatedx::end_bulk_insert"); + + if (bulk_insert.str && bulk_insert.length && !abort) + { + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + if (io->query(bulk_insert.str, bulk_insert.length)) + error= stash_remote_error(); + else + if (table->next_number_field) + update_auto_increment(); + } + + dynstr_free(&bulk_insert); + + DBUG_RETURN(my_errno= error); +} + + +/* + ha_federatedx::update_auto_increment + + This method ensures that last_insert_id() works properly. What it simply does + is calls last_insert_id() on the foreign database immediately after insert + (if the table has an auto_increment field) and sets the insert id via + thd->insert_id(ID)). +*/ +void ha_federatedx::update_auto_increment(void) +{ + THD *thd= current_thd; + DBUG_ENTER("ha_federatedx::update_auto_increment"); + + ha_federatedx::info(HA_STATUS_AUTO); + thd->first_successful_insert_id_in_cur_stmt= + stats.auto_increment_value; + DBUG_PRINT("info",("last_insert_id: %ld", (long) stats.auto_increment_value)); + + DBUG_VOID_RETURN; +} + +int ha_federatedx::optimize(THD* thd, HA_CHECK_OPT* check_opt) +{ + int error= 0; + char query_buffer[STRING_BUFFER_USUAL_SIZE]; + String query(query_buffer, sizeof(query_buffer), &my_charset_bin); + DBUG_ENTER("ha_federatedx::optimize"); + + query.length(0); + + query.set_charset(system_charset_info); + query.append(STRING_WITH_LEN("OPTIMIZE TABLE ")); + append_ident(&query, share->table_name, share->table_name_length, + ident_quote_char); + + DBUG_ASSERT(txn == get_txn(thd)); + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (io->query(query.ptr(), query.length())) + error= stash_remote_error(); + + DBUG_RETURN(error); +} + + +int ha_federatedx::repair(THD* thd, HA_CHECK_OPT* check_opt) +{ + int error= 0; + char query_buffer[STRING_BUFFER_USUAL_SIZE]; + String query(query_buffer, sizeof(query_buffer), &my_charset_bin); + DBUG_ENTER("ha_federatedx::repair"); + + query.length(0); + + query.set_charset(system_charset_info); + query.append(STRING_WITH_LEN("REPAIR TABLE ")); + append_ident(&query, share->table_name, share->table_name_length, + ident_quote_char); + if (check_opt->flags & T_QUICK) + query.append(STRING_WITH_LEN(" QUICK")); + if (check_opt->flags & T_EXTEND) + query.append(STRING_WITH_LEN(" EXTENDED")); + if (check_opt->sql_flags & TT_USEFRM) + query.append(STRING_WITH_LEN(" USE_FRM")); + + DBUG_ASSERT(txn == get_txn(thd)); + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (io->query(query.ptr(), query.length())) + error= stash_remote_error(); + + DBUG_RETURN(error); +} + + +/* + Yes, update_row() does what you expect, it updates a row. old_data will have + the previous row record in it, while new_data will have the newest data in + it. + + Keep in mind that the server can do updates based on ordering if an ORDER BY + clause was used. Consecutive ordering is not guaranteed. + Currently new_data will not have an updated auto_increament record, or + and updated timestamp field. You can do these for federatedx by doing these: + if (table->timestamp_on_update_now) + update_timestamp(new_row+table->timestamp_on_update_now-1); + if (table->next_number_field && record == table->record[0]) + update_auto_increment(); + + Called from sql_select.cc, sql_acl.cc, sql_update.cc, and sql_insert.cc. +*/ + +int ha_federatedx::update_row(const uchar *old_data, uchar *new_data) +{ + /* + This used to control how the query was built. If there was a + primary key, the query would be built such that there was a where + clause with only that column as the condition. This is flawed, + because if we have a multi-part primary key, it would only use the + first part! We don't need to do this anyway, because + read_range_first will retrieve the correct record, which is what + is used to build the WHERE clause. We can however use this to + append a LIMIT to the end if there is NOT a primary key. Why do + this? Because we only are updating one record, and LIMIT enforces + this. + */ + bool has_a_primary_key= test(table->s->primary_key != MAX_KEY); + + /* + buffers for following strings + */ + char field_value_buffer[STRING_BUFFER_USUAL_SIZE]; + char update_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + char where_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + + /* Work area for field values */ + String field_value(field_value_buffer, sizeof(field_value_buffer), + &my_charset_bin); + /* stores the update query */ + String update_string(update_buffer, + sizeof(update_buffer), + &my_charset_bin); + /* stores the WHERE clause */ + String where_string(where_buffer, + sizeof(where_buffer), + &my_charset_bin); + uchar *record= table->record[0]; + int error; + DBUG_ENTER("ha_federatedx::update_row"); + /* + set string lengths to 0 to avoid misc chars in string + */ + field_value.length(0); + update_string.length(0); + where_string.length(0); + + if (ignore_duplicates) + update_string.append(STRING_WITH_LEN("UPDATE IGNORE ")); + else + update_string.append(STRING_WITH_LEN("UPDATE ")); + append_ident(&update_string, share->table_name, + share->table_name_length, ident_quote_char); + update_string.append(STRING_WITH_LEN(" SET ")); + + /* + In this loop, we want to match column names to values being inserted + (while building INSERT statement). + + Iterate through table->field (new data) and share->old_field (old_data) + using the same index to create an SQL UPDATE statement. New data is + used to create SET field=value and old data is used to create WHERE + field=oldvalue + */ + + for (Field **field= table->field; *field; field++) + { + if (bitmap_is_set(table->write_set, (*field)->field_index)) + { + uint field_name_length= strlen((*field)->field_name); + append_ident(&update_string, (*field)->field_name, field_name_length, + ident_quote_char); + update_string.append(STRING_WITH_LEN(" = ")); + + if ((*field)->is_null()) + update_string.append(STRING_WITH_LEN(" NULL ")); + else + { + /* otherwise = */ + my_bitmap_map *old_map= tmp_use_all_columns(table, table->read_set); + bool needs_quote= (*field)->str_needs_quotes(); + (*field)->val_str(&field_value); + if (needs_quote) + update_string.append(value_quote_char); + field_value.print(&update_string); + if (needs_quote) + update_string.append(value_quote_char); + field_value.length(0); + tmp_restore_column_map(table->read_set, old_map); + } + update_string.append(STRING_WITH_LEN(", ")); + } + + if (bitmap_is_set(table->read_set, (*field)->field_index)) + { + uint field_name_length= strlen((*field)->field_name); + append_ident(&where_string, (*field)->field_name, field_name_length, + ident_quote_char); + if (field_in_record_is_null(table, *field, (char*) old_data)) + where_string.append(STRING_WITH_LEN(" IS NULL ")); + else + { + bool needs_quote= (*field)->str_needs_quotes(); + where_string.append(STRING_WITH_LEN(" = ")); + (*field)->val_str(&field_value, + (old_data + (*field)->offset(record))); + if (needs_quote) + where_string.append(value_quote_char); + field_value.print(&where_string); + if (needs_quote) + where_string.append(value_quote_char); + field_value.length(0); + } + where_string.append(STRING_WITH_LEN(" AND ")); + } + } + + /* Remove last ', '. This works as there must be at least on updated field */ + update_string.length(update_string.length() - sizeof_trailing_comma); + + if (where_string.length()) + { + /* chop off trailing AND */ + where_string.length(where_string.length() - sizeof_trailing_and); + update_string.append(STRING_WITH_LEN(" WHERE ")); + update_string.append(where_string); + } + + /* + If this table has not a primary key, then we could possibly + update multiple rows. We want to make sure to only update one! + */ + if (!has_a_primary_key) + update_string.append(STRING_WITH_LEN(" LIMIT 1")); + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (io->query(update_string.ptr(), update_string.length())) + { + DBUG_RETURN(stash_remote_error()); + } + DBUG_RETURN(0); +} + +/* + This will delete a row. 'buf' will contain a copy of the row to be =deleted. + The server will call this right after the current row has been called (from + either a previous rnd_next() or index call). + If you keep a pointer to the last row or can access a primary key it will + make doing the deletion quite a bit easier. + Keep in mind that the server does no guarentee consecutive deletions. + ORDER BY clauses can be used. + + Called in sql_acl.cc and sql_udf.cc to manage internal table information. + Called in sql_delete.cc, sql_insert.cc, and sql_select.cc. In sql_select + it is used for removing duplicates while in insert it is used for REPLACE + calls. +*/ + +int ha_federatedx::delete_row(const uchar *buf) +{ + char delete_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + char data_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + String delete_string(delete_buffer, sizeof(delete_buffer), &my_charset_bin); + String data_string(data_buffer, sizeof(data_buffer), &my_charset_bin); + uint found= 0; + int error; + DBUG_ENTER("ha_federatedx::delete_row"); + + delete_string.length(0); + delete_string.append(STRING_WITH_LEN("DELETE FROM ")); + append_ident(&delete_string, share->table_name, + share->table_name_length, ident_quote_char); + delete_string.append(STRING_WITH_LEN(" WHERE ")); + + for (Field **field= table->field; *field; field++) + { + Field *cur_field= *field; + found++; + if (bitmap_is_set(table->read_set, cur_field->field_index)) + { + append_ident(&delete_string, (*field)->field_name, + strlen((*field)->field_name), ident_quote_char); + data_string.length(0); + if (cur_field->is_null()) + { + delete_string.append(STRING_WITH_LEN(" IS NULL ")); + } + else + { + bool needs_quote= cur_field->str_needs_quotes(); + delete_string.append(STRING_WITH_LEN(" = ")); + cur_field->val_str(&data_string); + if (needs_quote) + delete_string.append(value_quote_char); + data_string.print(&delete_string); + if (needs_quote) + delete_string.append(value_quote_char); + } + delete_string.append(STRING_WITH_LEN(" AND ")); + } + } + + // Remove trailing AND + delete_string.length(delete_string.length() - sizeof_trailing_and); + if (!found) + delete_string.length(delete_string.length() - sizeof_trailing_where); + + delete_string.append(STRING_WITH_LEN(" LIMIT 1")); + DBUG_PRINT("info", + ("Delete sql: %s", delete_string.c_ptr_quick())); + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (io->query(delete_string.ptr(), delete_string.length())) + { + DBUG_RETURN(stash_remote_error()); + } + stats.deleted+= (ha_rows) io->affected_rows(); + stats.records-= (ha_rows) io->affected_rows(); + DBUG_PRINT("info", + ("rows deleted %ld rows deleted for all time %ld", + (long) io->affected_rows(), (long) stats.deleted)); + + DBUG_RETURN(0); +} + + +/* + Positions an index cursor to the index specified in the handle. Fetches the + row if available. If the key value is null, begin at the first key of the + index. This method, which is called in the case of an SQL statement having + a WHERE clause on a non-primary key index, simply calls index_read_idx. +*/ + +int ha_federatedx::index_read(uchar *buf, const uchar *key, + uint key_len, ha_rkey_function find_flag) +{ + DBUG_ENTER("ha_federatedx::index_read"); + + if (stored_result) + (void) free_result(); + DBUG_RETURN(index_read_idx_with_result_set(buf, active_index, key, + key_len, find_flag, + &stored_result)); +} + + +/* + Positions an index cursor to the index specified in key. Fetches the + row if any. This is only used to read whole keys. + + This method is called via index_read in the case of a WHERE clause using + a primary key index OR is called DIRECTLY when the WHERE clause + uses a PRIMARY KEY index. + + NOTES + This uses an internal result set that is deleted before function + returns. We need to be able to be callable from ha_rnd_pos() +*/ + +int ha_federatedx::index_read_idx(uchar *buf, uint index, const uchar *key, + uint key_len, enum ha_rkey_function find_flag) +{ + int retval; + FEDERATEDX_IO_RESULT *io_result; + DBUG_ENTER("ha_federatedx::index_read_idx"); + + if ((retval= index_read_idx_with_result_set(buf, index, key, + key_len, find_flag, + &io_result))) + DBUG_RETURN(retval); + /* io is correct, as index_read_idx_with_result_set was ok */ + io->free_result(io_result); + DBUG_RETURN(retval); +} + + +/* + Create result set for rows matching query and return first row + + RESULT + 0 ok In this case *result will contain the result set + table->status == 0 + # error In this case *result will contain 0 + table->status == STATUS_NOT_FOUND +*/ + +int ha_federatedx::index_read_idx_with_result_set(uchar *buf, uint index, + const uchar *key, + uint key_len, + ha_rkey_function find_flag, + FEDERATEDX_IO_RESULT **result) +{ + int retval; + char error_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + char index_value[STRING_BUFFER_USUAL_SIZE]; + char sql_query_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + String index_string(index_value, + sizeof(index_value), + &my_charset_bin); + String sql_query(sql_query_buffer, + sizeof(sql_query_buffer), + &my_charset_bin); + key_range range; + DBUG_ENTER("ha_federatedx::index_read_idx_with_result_set"); + + *result= 0; // In case of errors + index_string.length(0); + sql_query.length(0); + ha_statistic_increment(&SSV::ha_read_key_count); + + sql_query.append(share->select_query); + + range.key= key; + range.length= key_len; + range.flag= find_flag; + create_where_from_key(&index_string, + &table->key_info[index], + &range, + NULL, 0, 0); + sql_query.append(index_string); + + if ((retval= txn->acquire(share, TRUE, &io))) + DBUG_RETURN(retval); + + if (io->query(sql_query.ptr(), sql_query.length())) + { + my_sprintf(error_buffer, (error_buffer, "error: %d '%s'", + io->error_code(), io->error_str())); + retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE; + goto error; + } + if (!(*result= io->store_result())) + { + retval= HA_ERR_END_OF_FILE; + goto error; + } + if (!(retval= read_next(buf, *result))) + DBUG_RETURN(retval); + + io->free_result(*result); + *result= 0; + table->status= STATUS_NOT_FOUND; + DBUG_RETURN(retval); + +error: + table->status= STATUS_NOT_FOUND; + my_error(retval, MYF(0), error_buffer); + DBUG_RETURN(retval); +} + + +/* + This method is used exlusevely by filesort() to check if we + can create sorting buffers of necessary size. + If the handler returns more records that it declares + here server can just crash on filesort(). + We cannot guarantee that's not going to happen with + the FEDERATEDX engine, as we have records==0 always if the + client is a VIEW, and for the table the number of + records can inpredictably change during execution. + So we return maximum possible value here. +*/ + +ha_rows ha_federatedx::estimate_rows_upper_bound() +{ + return HA_POS_ERROR; +} + + +/* Initialized at each key walk (called multiple times unlike rnd_init()) */ + +int ha_federatedx::index_init(uint keynr, bool sorted) +{ + DBUG_ENTER("ha_federatedx::index_init"); + DBUG_PRINT("info", ("table: '%s' key: %u", table->s->table_name.str, keynr)); + active_index= keynr; + DBUG_RETURN(0); +} + + +/* + Read first range +*/ + +int ha_federatedx::read_range_first(const key_range *start_key, + const key_range *end_key, + bool eq_range_arg, bool sorted) +{ + char sql_query_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + int retval; + String sql_query(sql_query_buffer, + sizeof(sql_query_buffer), + &my_charset_bin); + DBUG_ENTER("ha_federatedx::read_range_first"); + + DBUG_ASSERT(!(start_key == NULL && end_key == NULL)); + + sql_query.length(0); + sql_query.append(share->select_query); + create_where_from_key(&sql_query, + &table->key_info[active_index], + start_key, end_key, 0, eq_range_arg); + + if ((retval= txn->acquire(share, TRUE, &io))) + DBUG_RETURN(retval); + + if (stored_result) + { + io->free_result(stored_result); + stored_result= 0; + } + + if (io->query(sql_query.ptr(), sql_query.length())) + { + retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE; + goto error; + } + sql_query.length(0); + + if (!(stored_result= io->store_result())) + { + retval= HA_ERR_END_OF_FILE; + goto error; + } + + retval= read_next(table->record[0], stored_result); + DBUG_RETURN(retval); + +error: + table->status= STATUS_NOT_FOUND; + DBUG_RETURN(retval); +} + + +int ha_federatedx::read_range_next() +{ + int retval; + DBUG_ENTER("ha_federatedx::read_range_next"); + retval= rnd_next(table->record[0]); + DBUG_RETURN(retval); +} + + +/* Used to read forward through the index. */ +int ha_federatedx::index_next(uchar *buf) +{ + DBUG_ENTER("ha_federatedx::index_next"); + ha_statistic_increment(&SSV::ha_read_next_count); + DBUG_RETURN(read_next(buf, stored_result)); +} + + +/* + rnd_init() is called when the system wants the storage engine to do a table + scan. + + This is the method that gets data for the SELECT calls. + + See the federatedx in the introduction at the top of this file to see when + rnd_init() is called. + + Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, + sql_table.cc, and sql_update.cc. +*/ + +int ha_federatedx::rnd_init(bool scan) +{ + DBUG_ENTER("ha_federatedx::rnd_init"); + /* + The use of the 'scan' flag is incredibly important for this handler + to work properly, especially with updates containing WHERE clauses + using indexed columns. + + When the initial query contains a WHERE clause of the query using an + indexed column, it's index_read_idx that selects the exact record from + the foreign database. + + When there is NO index in the query, either due to not having a WHERE + clause, or the WHERE clause is using columns that are not indexed, a + 'full table scan' done by rnd_init, which in this situation simply means + a 'select * from ...' on the foreign table. + + In other words, this 'scan' flag gives us the means to ensure that if + there is an index involved in the query, we want index_read_idx to + retrieve the exact record (scan flag is 0), and do not want rnd_init + to do a 'full table scan' and wipe out that result set. + + Prior to using this flag, the problem was most apparent with updates. + + An initial query like 'UPDATE tablename SET anything = whatever WHERE + indexedcol = someval', index_read_idx would get called, using a query + constructed with a WHERE clause built from the values of index ('indexcol' + in this case, having a value of 'someval'). mysql_store_result would + then get called (this would be the result set we want to use). + + After this rnd_init (from sql_update.cc) would be called, it would then + unecessarily call "select * from table" on the foreign table, then call + mysql_store_result, which would wipe out the correct previous result set + from the previous call of index_read_idx's that had the result set + containing the correct record, hence update the wrong row! + + */ + + if (scan) + { + int error; + + if ((error= txn->acquire(share, TRUE, &io))) + DBUG_RETURN(error); + + if (stored_result) + { + io->free_result(stored_result); + stored_result= 0; + } + + if (io->query(share->select_query, + strlen(share->select_query))) + goto error; + + stored_result= io->store_result(); + if (!stored_result) + goto error; + } + DBUG_RETURN(0); + +error: + DBUG_RETURN(stash_remote_error()); +} + + +int ha_federatedx::rnd_end() +{ + DBUG_ENTER("ha_federatedx::rnd_end"); + DBUG_RETURN(index_end()); +} + + +int ha_federatedx::free_result() +{ + int error; + DBUG_ASSERT(stored_result); + if ((error= txn->acquire(share, FALSE, &io))) + { + DBUG_ASSERT(0); // Fail when testing + return error; + } + io->free_result(stored_result); + stored_result= 0; + return 0; +} + +int ha_federatedx::index_end(void) +{ + int error= 0; + DBUG_ENTER("ha_federatedx::index_end"); + if (stored_result) + error= free_result(); + active_index= MAX_KEY; + DBUG_RETURN(error); +} + + +/* + This is called for each row of the table scan. When you run out of records + you should return HA_ERR_END_OF_FILE. Fill buff up with the row information. + The Field structure for the table is the key to getting data into buf + in a manner that will allow the server to understand it. + + Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, + sql_table.cc, and sql_update.cc. +*/ + +int ha_federatedx::rnd_next(uchar *buf) +{ + DBUG_ENTER("ha_federatedx::rnd_next"); + + if (stored_result == 0) + { + /* + Return value of rnd_init is not always checked (see records.cc), + so we can get here _even_ if there is _no_ pre-fetched result-set! + TODO: fix it. We can delete this in 5.1 when rnd_init() is checked. + */ + DBUG_RETURN(1); + } + DBUG_RETURN(read_next(buf, stored_result)); +} + + +/* + ha_federatedx::read_next + + reads from a result set and converts to mysql internal + format + + SYNOPSIS + field_in_record_is_null() + buf byte pointer to record + result mysql result set + + DESCRIPTION + This method is a wrapper method that reads one record from a result + set and converts it to the internal table format + + RETURN VALUE + 1 error + 0 no error +*/ + +int ha_federatedx::read_next(uchar *buf, FEDERATEDX_IO_RESULT *result) +{ + int retval; + FEDERATEDX_IO_ROW *row; + DBUG_ENTER("ha_federatedx::read_next"); + + table->status= STATUS_NOT_FOUND; // For easier return + + if ((retval= txn->acquire(share, TRUE, &io))) + DBUG_RETURN(retval); + + /* Fetch a row, insert it back in a row format. */ + if (!(row= io->fetch_row(result))) + DBUG_RETURN(HA_ERR_END_OF_FILE); + + if (!(retval= convert_row_to_internal_format(buf, row, result))) + table->status= 0; + + DBUG_RETURN(retval); +} + + +/* + store reference to current row so that we can later find it for + a re-read, update or delete. + + In case of federatedx, a reference is either a primary key or + the whole record. + + Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc. +*/ + +void ha_federatedx::position(const uchar *record) +{ + DBUG_ENTER("ha_federatedx::position"); + if (table->s->primary_key != MAX_KEY) + key_copy(ref, (uchar *)record, table->key_info + table->s->primary_key, + ref_length); + else + memcpy(ref, record, ref_length); + DBUG_VOID_RETURN; +} + + +/* + This is like rnd_next, but you are given a position to use to determine the + row. The position will be of the type that you stored in ref. + + This method is required for an ORDER BY + + Called from filesort.cc records.cc sql_insert.cc sql_select.cc sql_update.cc. +*/ + +int ha_federatedx::rnd_pos(uchar *buf, uchar *pos) +{ + int result; + DBUG_ENTER("ha_federatedx::rnd_pos"); + ha_statistic_increment(&SSV::ha_read_rnd_count); + if (table->s->primary_key != MAX_KEY) + { + /* We have a primary key, so use index_read_idx to find row */ + result= index_read_idx(buf, table->s->primary_key, pos, + ref_length, HA_READ_KEY_EXACT); + } + else + { + /* otherwise, get the old record ref as obtained in ::position */ + memcpy(buf, pos, ref_length); + result= 0; + } + table->status= result ? STATUS_NOT_FOUND : 0; + DBUG_RETURN(result); +} + + +/* + ::info() is used to return information to the optimizer. + Currently this table handler doesn't implement most of the fields + really needed. SHOW also makes use of this data + Another note, you will probably want to have the following in your + code: + if (records < 2) + records = 2; + The reason is that the server will optimize for cases of only a single + record. If in a table scan you don't know the number of records + it will probably be better to set records to two so you can return + as many records as you need. + Along with records a few more variables you may wish to set are: + records + deleted + data_file_length + index_file_length + delete_length + check_time + Take a look at the public variables in handler.h for more information. + + Called in: + filesort.cc + ha_heap.cc + item_sum.cc + opt_sum.cc + sql_delete.cc + sql_delete.cc + sql_derived.cc + sql_select.cc + sql_select.cc + sql_select.cc + sql_select.cc + sql_select.cc + sql_show.cc + sql_show.cc + sql_show.cc + sql_show.cc + sql_table.cc + sql_union.cc + sql_update.cc + +*/ + +int ha_federatedx::info(uint flag) +{ + char error_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + uint error_code; + federatedx_io *tmp_io= 0; + DBUG_ENTER("ha_federatedx::info"); + + error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE; + + /* we want not to show table status if not needed to do so */ + if (flag & (HA_STATUS_VARIABLE | HA_STATUS_CONST | HA_STATUS_AUTO)) + { + if ((error_code= txn->acquire(share, TRUE, &tmp_io))) + goto fail; + } + + if (flag & (HA_STATUS_VARIABLE | HA_STATUS_CONST)) + { + /* + size of IO operations (This is based on a good guess, no high science + involved) + */ + if (flag & HA_STATUS_CONST) + stats.block_size= 4096; + + if (tmp_io->table_metadata(&stats, share->table_name, + share->table_name_length, flag)) + goto error; + } + + if (flag & HA_STATUS_AUTO) + stats.auto_increment_value= tmp_io->last_insert_id(); + + /* + If ::info created it's own transaction, close it. This happens in case + of show table status; + */ + txn->release(&tmp_io); + + DBUG_RETURN(0); + +error: + if (tmp_io) + { + my_sprintf(error_buffer, (error_buffer, ": %d : %s", + tmp_io->error_code(), tmp_io->error_str())); + my_error(error_code, MYF(0), error_buffer); + } + else + if (remote_error_number != -1 /* error already reported */) + { + error_code= remote_error_number; + my_error(error_code, MYF(0), ER(error_code)); + } +fail: + txn->release(&tmp_io); + DBUG_RETURN(error_code); +} + + +/** + @brief Handles extra signals from MySQL server + + @param[in] operation Hint for storage engine + + @return Operation Status + @retval 0 OK + */ +int ha_federatedx::extra(ha_extra_function operation) +{ + DBUG_ENTER("ha_federatedx::extra"); + switch (operation) { + case HA_EXTRA_IGNORE_DUP_KEY: + ignore_duplicates= TRUE; + break; + case HA_EXTRA_NO_IGNORE_DUP_KEY: + insert_dup_update= FALSE; + ignore_duplicates= FALSE; + break; + case HA_EXTRA_WRITE_CAN_REPLACE: + replace_duplicates= TRUE; + break; + case HA_EXTRA_WRITE_CANNOT_REPLACE: + /* + We use this flag to ensure that we do not create an "INSERT IGNORE" + statement when inserting new rows into the remote table. + */ + replace_duplicates= FALSE; + break; + case HA_EXTRA_INSERT_WITH_UPDATE: + insert_dup_update= TRUE; + break; + default: + /* do nothing */ + DBUG_PRINT("info",("unhandled operation: %d", (uint) operation)); + } + DBUG_RETURN(0); +} + + +/** + @brief Reset state of file to after 'open'. + + @detail This function is called after every statement for all tables + used by that statement. + + @return Operation status + @retval 0 OK +*/ + +int ha_federatedx::reset(void) +{ + insert_dup_update= FALSE; + ignore_duplicates= FALSE; + replace_duplicates= FALSE; + return 0; +} + + +/* + Used to delete all rows in a table. Both for cases of truncate and + for cases where the optimizer realizes that all rows will be + removed as a result of a SQL statement. + + Called from item_sum.cc by Item_func_group_concat::clear(), + Item_sum_count_distinct::clear(), and Item_func_group_concat::clear(). + Called from sql_delete.cc by mysql_delete(). + Called from sql_select.cc by JOIN::reinit(). + Called from sql_union.cc by st_select_lex_unit::exec(). +*/ + +int ha_federatedx::delete_all_rows() +{ + char query_buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + String query(query_buffer, sizeof(query_buffer), &my_charset_bin); + int error; + DBUG_ENTER("ha_federatedx::delete_all_rows"); + + query.length(0); + + query.set_charset(system_charset_info); + query.append(STRING_WITH_LEN("TRUNCATE ")); + append_ident(&query, share->table_name, share->table_name_length, + ident_quote_char); + + /* no need for savepoint in autocommit mode */ + if (!(ha_thd()->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) + txn->stmt_autocommit(); + + /* + TRUNCATE won't return anything in mysql_affected_rows + */ + + if ((error= txn->acquire(share, FALSE, &io))) + DBUG_RETURN(error); + + if (io->query(query.ptr(), query.length())) + { + DBUG_RETURN(stash_remote_error()); + } + stats.deleted+= stats.records; + stats.records= 0; + DBUG_RETURN(0); +} + + +/* + The idea with handler::store_lock() is the following: + + The statement decided which locks we should need for the table + for updates/deletes/inserts we get WRITE locks, for SELECT... we get + read locks. + + Before adding the lock into the table lock handler (see thr_lock.c) + mysqld calls store lock with the requested locks. Store lock can now + modify a write lock to a read lock (or some other lock), ignore the + lock (if we don't want to use MySQL table locks at all) or add locks + for many tables (like we do when we are using a MERGE handler). + + Berkeley DB for federatedx changes all WRITE locks to TL_WRITE_ALLOW_WRITE + (which signals that we are doing WRITES, but we are still allowing other + reader's and writer's. + + When releasing locks, store_lock() are also called. In this case one + usually doesn't have to do anything. + + In some exceptional cases MySQL may send a request for a TL_IGNORE; + This means that we are requesting the same lock as last time and this + should also be ignored. (This may happen when someone does a flush + table when we have opened a part of the tables, in which case mysqld + closes and reopens the tables and tries to get the same locks at last + time). In the future we will probably try to remove this. + + Called from lock.cc by get_lock_data(). +*/ + +THR_LOCK_DATA **ha_federatedx::store_lock(THD *thd, + THR_LOCK_DATA **to, + enum thr_lock_type lock_type) +{ + DBUG_ENTER("ha_federatedx::store_lock"); + if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK) + { + /* + Here is where we get into the guts of a row level lock. + If TL_UNLOCK is set + If we are not doing a LOCK TABLE or DISCARD/IMPORT + TABLESPACE, then allow multiple writers + */ + + if ((lock_type >= TL_WRITE_CONCURRENT_INSERT && + lock_type <= TL_WRITE) && !thd->in_lock_tables) + lock_type= TL_WRITE_ALLOW_WRITE; + + /* + In queries of type INSERT INTO t1 SELECT ... FROM t2 ... + MySQL would use the lock TL_READ_NO_INSERT on t2, and that + would conflict with TL_WRITE_ALLOW_WRITE, blocking all inserts + to t2. Convert the lock to a normal read lock to allow + concurrent inserts to t2. + */ + + if (lock_type == TL_READ_NO_INSERT && !thd->in_lock_tables) + lock_type= TL_READ; + + lock.type= lock_type; + } + + *to++= &lock; + + DBUG_RETURN(to); +} + + +static int test_connection(MYSQL_THD thd, federatedx_io *io, + FEDERATEDX_SHARE *share) +{ + char buffer[FEDERATEDX_QUERY_BUFFER_SIZE]; + String str(buffer, sizeof(buffer), &my_charset_bin); + FEDERATEDX_IO_RESULT *resultset= NULL; + int retval; + + str.length(0); + str.append(STRING_WITH_LEN("SELECT * FROM ")); + append_identifier(thd, &str, share->table_name, + share->table_name_length); + str.append(STRING_WITH_LEN(" WHERE 1=0")); + + if ((retval= io->query(str.ptr(), str.length()))) + { + my_sprintf(buffer, (buffer, + "database: '%s' username: '%s' hostname: '%s'", + share->database, share->username, share->hostname)); + DBUG_PRINT("info", ("error-code: %d", io->error_code())); + my_error(ER_CANT_CREATE_FEDERATED_TABLE, MYF(0), buffer); + } + else + resultset= io->store_result(); + + io->free_result(resultset); + + return retval; +} + +/* + create() does nothing, since we have no local setup of our own. + FUTURE: We should potentially connect to the foreign database and +*/ + +int ha_federatedx::create(const char *name, TABLE *table_arg, + HA_CREATE_INFO *create_info) +{ + int retval; + THD *thd= current_thd; + FEDERATEDX_SHARE tmp_share; // Only a temporary share, to test the url + federatedx_txn *tmp_txn; + federatedx_io *tmp_io= NULL; + DBUG_ENTER("ha_federatedx::create"); + + if ((retval= parse_url(thd->mem_root, &tmp_share, table_arg, 1))) + goto error; + + /* loopback socket connections hang due to LOCK_open mutex */ + if ((!tmp_share.hostname || !strcmp(tmp_share.hostname,my_localhost)) && + !tmp_share.port) + goto error; + + /* + If possible, we try to use an existing network connection to + the remote server. To ensure that no new FEDERATEDX_SERVER + instance is created, we pass NULL in get_server() TABLE arg. + */ + pthread_mutex_lock(&federatedx_mutex); + tmp_share.s= get_server(&tmp_share, NULL); + pthread_mutex_unlock(&federatedx_mutex); + + if (tmp_share.s) + { + tmp_txn= get_txn(thd); + if (!(retval= tmp_txn->acquire(&tmp_share, TRUE, &tmp_io))) + { + retval= test_connection(thd, tmp_io, &tmp_share); + tmp_txn->release(&tmp_io); + } + free_server(tmp_txn, tmp_share.s); + } + else + { + FEDERATEDX_SERVER server; + +#ifdef NOT_YET + /* + Bug#25679 + Ensure that we do not hold the LOCK_open mutex while attempting + to establish FederatedX connection to guard against a trivial + Denial of Service scenerio. + */ + safe_mutex_assert_not_owner(&LOCK_open); +#endif + + fill_server(thd->mem_root, &server, &tmp_share, create_info->table_charset); + +#ifndef DBUG_OFF + pthread_mutex_init(&server.mutex, MY_MUTEX_INIT_FAST); + pthread_mutex_lock(&server.mutex); +#endif + + tmp_io= federatedx_io::construct(thd->mem_root, &server); + + retval= test_connection(thd, tmp_io, &tmp_share); + +#ifndef DBUG_OFF + pthread_mutex_unlock(&server.mutex); + pthread_mutex_destroy(&server.mutex); +#endif + + delete tmp_io; + } + +error: + DBUG_RETURN(retval); + +} + + +int ha_federatedx::stash_remote_error() +{ + DBUG_ENTER("ha_federatedx::stash_remote_error()"); + if (!io) + DBUG_RETURN(remote_error_number); + remote_error_number= io->error_code(); + strmake(remote_error_buf, io->error_str(), sizeof(remote_error_buf)-1); + if (remote_error_number == ER_DUP_ENTRY || + remote_error_number == ER_DUP_KEY) + DBUG_RETURN(HA_ERR_FOUND_DUPP_KEY); + DBUG_RETURN(HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM); +} + + +bool ha_federatedx::get_error_message(int error, String* buf) +{ + DBUG_ENTER("ha_federatedx::get_error_message"); + DBUG_PRINT("enter", ("error: %d", error)); + if (error == HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM) + { + buf->append(STRING_WITH_LEN("Error on remote system: ")); + buf->qs_append(remote_error_number); + buf->append(STRING_WITH_LEN(": ")); + buf->append(remote_error_buf); + + remote_error_number= 0; + remote_error_buf[0]= '\0'; + } + DBUG_PRINT("exit", ("message: %s", buf->ptr())); + DBUG_RETURN(FALSE); +} + + +int ha_federatedx::start_stmt(MYSQL_THD thd, thr_lock_type lock_type) +{ + DBUG_ENTER("ha_federatedx::start_stmt"); + DBUG_ASSERT(txn == get_txn(thd)); + + if (!txn->in_transaction()) + { + txn->stmt_begin(); + trans_register_ha(thd, FALSE, ht); + } + DBUG_RETURN(0); +} + + +int ha_federatedx::external_lock(MYSQL_THD thd, int lock_type) +{ + int error= 0; + DBUG_ENTER("ha_federatedx::external_lock"); + + if (lock_type == F_UNLCK) + txn->release(&io); + else + { + txn= get_txn(thd); + if (!(error= txn->acquire(share, lock_type == F_RDLCK, &io)) && + (lock_type == F_WRLCK || !io->is_autocommit())) + { + if (!thd_test_options(thd, (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) + { + txn->stmt_begin(); + trans_register_ha(thd, FALSE, ht); + } + else + { + txn->txn_begin(); + trans_register_ha(thd, TRUE, ht); + } + } + } + + DBUG_RETURN(error); +} + + +int ha_federatedx::savepoint_set(handlerton *hton, MYSQL_THD thd, void *sv) +{ + int error= 0; + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + DBUG_ENTER("ha_federatedx::savepoint_set"); + + if (txn && txn->has_connections()) + { + if (txn->txn_begin()) + trans_register_ha(thd, TRUE, hton); + + txn->sp_acquire((ulong *) sv); + + DBUG_ASSERT(1 < *(ulong *) sv); + } + + DBUG_RETURN(error); +} + + +int ha_federatedx::savepoint_rollback(handlerton *hton, MYSQL_THD thd, void *sv) + { + int error= 0; + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + DBUG_ENTER("ha_federatedx::savepoint_rollback"); + + if (txn) + error= txn->sp_rollback((ulong *) sv); + + DBUG_RETURN(error); +} + + +int ha_federatedx::savepoint_release(handlerton *hton, MYSQL_THD thd, void *sv) +{ + int error= 0; + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + DBUG_ENTER("ha_federatedx::savepoint_release"); + + if (txn) + error= txn->sp_release((ulong *) sv); + + DBUG_RETURN(error); +} + + +int ha_federatedx::commit(handlerton *hton, MYSQL_THD thd, bool all) +{ + int return_val; + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + DBUG_ENTER("ha_federatedx::commit"); + + if (all) + return_val= txn->txn_commit(); + else + return_val= txn->stmt_commit(); + + DBUG_PRINT("info", ("error val: %d", return_val)); + DBUG_RETURN(return_val); +} + + +int ha_federatedx::rollback(handlerton *hton, MYSQL_THD thd, bool all) +{ + int return_val; + federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton); + DBUG_ENTER("ha_federatedx::rollback"); + + if (all) + return_val= txn->txn_rollback(); + else + return_val= txn->stmt_rollback(); + + DBUG_PRINT("info", ("error val: %d", return_val)); + DBUG_RETURN(return_val); +} + +struct st_mysql_storage_engine federatedx_storage_engine= +{ MYSQL_HANDLERTON_INTERFACE_VERSION }; + +mysql_declare_plugin(federated) +{ + MYSQL_STORAGE_ENGINE_PLUGIN, + &federatedx_storage_engine, + "FEDERATED", + "Patrick Galbraith", + "FederatedX pluggable storage engine", + PLUGIN_LICENSE_GPL, + federatedx_db_init, /* Plugin Init */ + federatedx_done, /* Plugin Deinit */ + 0x0100 /* 1.0 */, + NULL, /* status variables */ + NULL, /* system variables */ + NULL /* config options */ +} +mysql_declare_plugin_end; === added file 'storage/federatedx/ha_federatedx.h' --- a/storage/federatedx/ha_federatedx.h 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/ha_federatedx.h 2009-10-30 18:50:56 +0000 @@ -0,0 +1,446 @@ +/* +Copyright (c) 2008, Patrick Galbraith +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above +copyright notice, this list of conditions and the following disclaimer +in the documentation and/or other materials provided with the +distribution. + + * Neither the name of Patrick Galbraith nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + + +class federatedx_io; + +/* + FEDERATEDX_SERVER will eventually be a structure that will be shared among + all FEDERATEDX_SHARE instances so that the federated server can minimise + the number of open connections. This will eventually lead to the support + of reliable XA federated tables. +*/ +typedef struct st_fedrated_server { + MEM_ROOT mem_root; + uint use_count, io_count; + + uchar *key; + uint key_length; + + const char *scheme; + const char *hostname; + const char *username; + const char *password; + const char *database; + const char *socket; + ushort port; + + const char *csname; + + pthread_mutex_t mutex; + federatedx_io *idle_list; +} FEDERATEDX_SERVER; + +/* + Please read ha_exmple.cc before reading this file. + Please keep in mind that the federatedx storage engine implements all methods + that are required to be implemented. handler.h has a full list of methods + that you can implement. +*/ + +#ifdef USE_PRAGMA_INTERFACE +#pragma interface /* gcc class implementation */ +#endif + +#include <mysql.h> + +/* + handler::print_error has a case statement for error numbers. + This value is (10000) is far out of range and will envoke the + default: case. + (Current error range is 120-159 from include/my_base.h) +*/ +#define HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM 10000 + +#define FEDERATEDX_QUERY_BUFFER_SIZE STRING_BUFFER_USUAL_SIZE * 5 +#define FEDERATEDX_RECORDS_IN_RANGE 2 +#define FEDERATEDX_MAX_KEY_LENGTH 3500 // Same as innodb + +/* + FEDERATEDX_SHARE is a structure that will be shared amoung all open handlers + The example implements the minimum of what you will probably need. +*/ +typedef struct st_federatedx_share { + MEM_ROOT mem_root; + + bool parsed; + /* this key is unique db/tablename */ + const char *share_key; + /* + the primary select query to be used in rnd_init + */ + char *select_query; + /* + remote host info, parse_url supplies + */ + char *server_name; + char *connection_string; + char *scheme; + char *hostname; + char *username; + char *password; + char *database; + char *table_name; + char *table; + char *socket; + char *sport; + int share_key_length; + ushort port; + + uint table_name_length, server_name_length, connect_string_length; + uint use_count; + THR_LOCK lock; + FEDERATEDX_SERVER *s; +} FEDERATEDX_SHARE; + + +typedef struct st_federatedx_result FEDERATEDX_IO_RESULT; +typedef struct st_federatedx_row FEDERATEDX_IO_ROW; +typedef ptrdiff_t FEDERATEDX_IO_OFFSET; + +class federatedx_io +{ + friend class federatedx_txn; + FEDERATEDX_SERVER * const server; + federatedx_io **owner_ptr; + federatedx_io *txn_next; + federatedx_io *idle_next; + bool active; /* currently participating in a transaction */ + bool busy; /* in use by a ha_federated instance */ + bool readonly;/* indicates that no updates have occurred */ + +protected: + void set_active(bool new_active) + { active= new_active; } +public: + federatedx_io(FEDERATEDX_SERVER *); + virtual ~federatedx_io(); + + bool is_readonly() const { return readonly; } + bool is_active() const { return active; } + + const char * get_charsetname() const + { return server->csname ? server->csname : "latin1"; } + + const char * get_hostname() const { return server->hostname; } + const char * get_username() const { return server->username; } + const char * get_password() const { return server->password; } + const char * get_database() const { return server->database; } + ushort get_port() const { return server->port; } + const char * get_socket() const { return server->socket; } + + static bool handles_scheme(const char *scheme); + static federatedx_io *construct(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server); + + static void *operator new(size_t size, MEM_ROOT *mem_root) throw () + { return alloc_root(mem_root, size); } + static void operator delete(void *ptr, size_t size) + { TRASH(ptr, size); } + + virtual int query(const char *buffer, uint length)=0; + virtual FEDERATEDX_IO_RESULT *store_result()=0; + + virtual size_t max_query_size() const=0; + + virtual my_ulonglong affected_rows() const=0; + virtual my_ulonglong last_insert_id() const=0; + + virtual int error_code()=0; + virtual const char *error_str()=0; + + virtual void reset()=0; + virtual int commit()=0; + virtual int rollback()=0; + + virtual int savepoint_set(ulong sp)=0; + virtual ulong savepoint_release(ulong sp)=0; + virtual ulong savepoint_rollback(ulong sp)=0; + virtual void savepoint_restrict(ulong sp)=0; + + virtual ulong last_savepoint() const=0; + virtual ulong actual_savepoint() const=0; + virtual bool is_autocommit() const=0; + + virtual bool table_metadata(ha_statistics *stats, const char *table_name, + uint table_name_length, uint flag) = 0; + + /* resultset operations */ + + virtual void free_result(FEDERATEDX_IO_RESULT *io_result)=0; + virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result)=0; + virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result)=0; + virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result)=0; + virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result)=0; + virtual const char *get_column_data(FEDERATEDX_IO_ROW *row, + unsigned int column)=0; + virtual bool is_column_null(const FEDERATEDX_IO_ROW *row, + unsigned int column) const=0; +}; + + +class federatedx_txn +{ + federatedx_io *txn_list; + ulong savepoint_level; + ulong savepoint_stmt; + ulong savepoint_next; + + void release_scan(); +public: + federatedx_txn(); + ~federatedx_txn(); + + bool has_connections() const { return txn_list != NULL; } + bool in_transaction() const { return savepoint_next != 0; } + int acquire(FEDERATEDX_SHARE *share, bool readonly, federatedx_io **io); + void release(federatedx_io **io); + void close(FEDERATEDX_SERVER *); + + bool txn_begin(); + int txn_commit(); + int txn_rollback(); + + bool sp_acquire(ulong *save); + int sp_rollback(ulong *save); + int sp_release(ulong *save); + + bool stmt_begin(); + int stmt_commit(); + int stmt_rollback(); + void stmt_autocommit(); +}; + + +/* + Class definition for the storage engine +*/ +class ha_federatedx: public handler +{ + friend int federatedx_db_init(void *p); + + THR_LOCK_DATA lock; /* MySQL lock */ + FEDERATEDX_SHARE *share; /* Shared lock info */ + federatedx_txn *txn; + federatedx_io *io; + FEDERATEDX_IO_RESULT *stored_result; + uint fetch_num; // stores the fetch num + FEDERATEDX_IO_OFFSET current_position; // Current position used by ::position() + int remote_error_number; + char remote_error_buf[FEDERATEDX_QUERY_BUFFER_SIZE]; + bool ignore_duplicates, replace_duplicates; + bool insert_dup_update; + DYNAMIC_STRING bulk_insert; + +private: + /* + return 0 on success + return errorcode otherwise + */ + uint convert_row_to_internal_format(uchar *buf, FEDERATEDX_IO_ROW *row, + FEDERATEDX_IO_RESULT *result); + bool create_where_from_key(String *to, KEY *key_info, + const key_range *start_key, + const key_range *end_key, + bool records_in_range, bool eq_range); + int stash_remote_error(); + + federatedx_txn *get_txn(THD *thd, bool no_create= FALSE); + + static int disconnect(handlerton *hton, MYSQL_THD thd); + static int savepoint_set(handlerton *hton, MYSQL_THD thd, void *sv); + static int savepoint_rollback(handlerton *hton, MYSQL_THD thd, void *sv); + static int savepoint_release(handlerton *hton, MYSQL_THD thd, void *sv); + static int commit(handlerton *hton, MYSQL_THD thd, bool all); + static int rollback(handlerton *hton, MYSQL_THD thd, bool all); + + bool append_stmt_insert(String *query); + + int read_next(uchar *buf, FEDERATEDX_IO_RESULT *result); + int index_read_idx_with_result_set(uchar *buf, uint index, + const uchar *key, + uint key_len, + ha_rkey_function find_flag, + FEDERATEDX_IO_RESULT **result); + int real_query(const char *query, uint length); + int real_connect(FEDERATEDX_SHARE *my_share, uint create_flag); +public: + ha_federatedx(handlerton *hton, TABLE_SHARE *table_arg); + ~ha_federatedx() {} + /* The name that will be used for display purposes */ + const char *table_type() const { return "FEDERATED"; } + /* + The name of the index type that will be used for display + don't implement this method unless you really have indexes + */ + // perhaps get index type + const char *index_type(uint inx) { return "REMOTE"; } + const char **bas_ext() const; + /* + This is a list of flags that says what the storage engine + implements. The current table flags are documented in + handler.h + */ + ulonglong table_flags() const + { + /* fix server to be able to get remote server table flags */ + return (HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED + | HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_CAN_INDEX_BLOBS | + HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE | + HA_NO_PREFIX_CHAR_KEYS | HA_PRIMARY_KEY_REQUIRED_FOR_DELETE | + HA_PARTIAL_COLUMN_READ | HA_NULL_IN_KEY); + } + /* + This is a bitmap of flags that says how the storage engine + implements indexes. The current index flags are documented in + handler.h. If you do not implement indexes, just return zero + here. + + part is the key part to check. First key part is 0 + If all_parts it's set, MySQL want to know the flags for the combined + index up to and including 'part'. + */ + /* fix server to be able to get remote server index flags */ + ulong index_flags(uint inx, uint part, bool all_parts) const + { + return (HA_READ_NEXT | HA_READ_RANGE | HA_READ_AFTER_KEY); + } + uint max_supported_record_length() const { return HA_MAX_REC_LENGTH; } + uint max_supported_keys() const { return MAX_KEY; } + uint max_supported_key_parts() const { return MAX_REF_PARTS; } + uint max_supported_key_length() const { return FEDERATEDX_MAX_KEY_LENGTH; } + uint max_supported_key_part_length() const { return FEDERATEDX_MAX_KEY_LENGTH; } + /* + Called in test_quick_select to determine if indexes should be used. + Normally, we need to know number of blocks . For federatedx we need to + know number of blocks on remote side, and number of packets and blocks + on the network side (?) + Talk to Kostja about this - how to get the + number of rows * ... + disk scan time on other side (block size, size of the row) + network time ... + The reason for "records * 1000" is that such a large number forces + this to use indexes " + */ + double scan_time() + { + DBUG_PRINT("info", ("records %lu", (ulong) stats.records)); + return (double)(stats.records*1000); + } + /* + The next method will never be called if you do not implement indexes. + */ + double read_time(uint index, uint ranges, ha_rows rows) + { + /* + Per Brian, this number is bugus, but this method must be implemented, + and at a later date, he intends to document this issue for handler code + */ + return (double) rows / 20.0+1; + } + + const key_map *keys_to_use_for_scanning() { return &key_map_full; } + /* + Everything below are methods that we implment in ha_federatedx.cc. + + Most of these methods are not obligatory, skip them and + MySQL will treat them as not implemented + */ + int open(const char *name, int mode, uint test_if_locked); // required + int close(void); // required + + void start_bulk_insert(ha_rows rows); + int end_bulk_insert(bool abort); + int write_row(uchar *buf); + int update_row(const uchar *old_data, uchar *new_data); + int delete_row(const uchar *buf); + int index_init(uint keynr, bool sorted); + ha_rows estimate_rows_upper_bound(); + int index_read(uchar *buf, const uchar *key, + uint key_len, enum ha_rkey_function find_flag); + int index_read_idx(uchar *buf, uint idx, const uchar *key, + uint key_len, enum ha_rkey_function find_flag); + int index_next(uchar *buf); + int index_end(); + int read_range_first(const key_range *start_key, + const key_range *end_key, + bool eq_range, bool sorted); + int read_range_next(); + /* + unlike index_init(), rnd_init() can be called two times + without rnd_end() in between (it only makes sense if scan=1). + then the second call should prepare for the new table scan + (e.g if rnd_init allocates the cursor, second call should + position it to the start of the table, no need to deallocate + and allocate it again + */ + int rnd_init(bool scan); //required + int rnd_end(); + int rnd_next(uchar *buf); //required + int rnd_pos(uchar *buf, uchar *pos); //required + void position(const uchar *record); //required + int info(uint); //required + int extra(ha_extra_function operation); + + void update_auto_increment(void); + int repair(THD* thd, HA_CHECK_OPT* check_opt); + int optimize(THD* thd, HA_CHECK_OPT* check_opt); + + int delete_all_rows(void); + int create(const char *name, TABLE *form, + HA_CREATE_INFO *create_info); //required + ha_rows records_in_range(uint inx, key_range *start_key, + key_range *end_key); + uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; } + + THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to, + enum thr_lock_type lock_type); //required + bool get_error_message(int error, String *buf); + int start_stmt(THD *thd, thr_lock_type lock_type); + int external_lock(THD *thd, int lock_type); + int reset(void); + int free_result(void); +}; + +extern const char ident_quote_char; // Character for quoting + // identifiers +extern const char value_quote_char; // Character for quoting + // literals + +extern bool append_ident(String *string, const char *name, uint length, + const char quote_char); + + +extern federatedx_io *instantiate_io_mysql(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server); +extern federatedx_io *instantiate_io_null(MEM_ROOT *server_root, + FEDERATEDX_SERVER *server); === added file 'storage/federatedx/plug.in' --- a/storage/federatedx/plug.in 1970-01-01 00:00:00 +0000 +++ b/storage/federatedx/plug.in 2009-10-30 18:50:56 +0000 @@ -0,0 +1,5 @@ +MYSQL_STORAGE_ENGINE(federated,,[FederatedX Storage Engine], + [FederatedX Storage Engine], [max,max-no-ndb]) +MYSQL_PLUGIN_DYNAMIC(federated, [ha_federatedx.la]) +MYSQL_PLUGIN_STATIC(federated, [libfederatedx.a]) +MYSQL_PLUGIN_DEPENDS_ON_MYSQL_INTERNALS(federated, [ha_federatedx.cc]) === modified file 'storage/pbxt/src/cache_xt.cc' --- a/storage/pbxt/src/cache_xt.cc 2009-08-17 11:12:36 +0000 +++ b/storage/pbxt/src/cache_xt.cc 2009-10-30 18:50:56 +0000 @@ -374,7 +374,7 @@ xtPublic void xt_ind_release_handle(XTIn { DcHandleSlotPtr hs; XTIndBlockPtr block = NULL; - u_int hash_idx = NULL; + u_int hash_idx = 0; DcSegmentPtr seg = NULL; XTIndBlockPtr xblock; @@ -1379,7 +1379,7 @@ xtPublic xtBool xt_ind_fetch(XTOpenTable ASSERT_NS(iref->ir_xlock == 2); #endif if (!(block = ind_cac_fetch(ot, ind, address, &seg, TRUE))) - return NULL; + return 0; branch_size = XT_GET_DISK_2(((XTIdxBranchDPtr) block->cb_data)->tb_size_2); if (XT_GET_INDEX_BLOCK_LEN(branch_size) < 2 || XT_GET_INDEX_BLOCK_LEN(branch_size) > XT_INDEX_PAGE_SIZE) { === modified file 'storage/xtradb/include/buf0buf.ic' --- a/storage/xtradb/include/buf0buf.ic 2009-06-25 01:43:25 +0000 +++ b/storage/xtradb/include/buf0buf.ic 2009-10-30 18:50:56 +0000 @@ -1056,7 +1056,7 @@ buf_page_release( buf_block_t* block, /* in: buffer block */ ulint rw_latch, /* in: RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH */ - mtr_t* mtr) /* in: mtr */ + mtr_t* mtr __attribute__((unused))) /* in: mtr */ { ut_ad(block);

1 0