220
int thd_in_lock_tables(const THD *thd)
222
return test(thd->in_lock_tables);
227
int thd_tablespace_op(const THD *thd)
229
return test(thd->tablespace_op);
234
const char *set_thd_proc_info(THD *thd, const char *info,
235
const char *calling_function __attribute__((unused)),
236
const char *calling_file __attribute__((unused)),
237
const unsigned int calling_line __attribute__((unused)))
239
const char *old_info= thd->get_proc_info();
240
thd->set_proc_info(info);
245
void **thd_ha_data(const THD *thd, const struct handlerton *hton)
247
return (void **) &thd->ha_data[hton->slot].ha_ptr;
251
int64_t thd_test_options(const THD *thd, int64_t test_options)
253
return thd->options & test_options;
257
int thd_sql_command(const THD *thd)
259
return (int) thd->lex->sql_command;
263
int thd_tx_isolation(const THD *thd)
265
return (int) thd->variables.tx_isolation;
269
void thd_inc_row_count(THD *thd)
275
Clear this diagnostics area.
277
Normally called at the end of a statement.
281
Diagnostics_area::reset_diagnostics_area()
283
can_overwrite_status= false;
284
/** Don't take chances in production */
290
m_total_warn_count= 0;
292
/** Tiny reset in debug mode to see garbage right away */
298
Set OK status -- ends commands that do not return a
299
result set, e.g. INSERT/UPDATE/DELETE.
303
Diagnostics_area::set_ok_status(THD *thd, ha_rows affected_rows_arg,
304
uint64_t last_insert_id_arg,
305
const char *message_arg)
309
In production, refuse to overwrite an error or a custom response
312
if (is_error() || is_disabled())
314
/** Only allowed to report success if has not yet reported an error */
316
m_server_status= thd->server_status;
317
m_total_warn_count= thd->total_warn_count;
318
m_affected_rows= affected_rows_arg;
319
m_last_insert_id= last_insert_id_arg;
321
strmake(m_message, message_arg, sizeof(m_message) - 1);
333
Diagnostics_area::set_eof_status(THD *thd)
335
/** Only allowed to report eof if has not yet reported an error */
339
In production, refuse to overwrite an error or a custom response
342
if (is_error() || is_disabled())
345
m_server_status= thd->server_status;
347
If inside a stored procedure, do not return the total
348
number of warnings, since they are not available to the client
351
m_total_warn_count= thd->total_warn_count;
361
Diagnostics_area::set_error_status(THD *thd __attribute__((unused)),
362
uint32_t sql_errno_arg,
363
const char *message_arg)
366
Only allowed to report error if has not yet reported a success
367
The only exception is when we flush the message to the client,
368
an error can happen during the flush.
370
assert(! is_set() || can_overwrite_status);
372
In production, refuse to overwrite a custom response with an
378
m_sql_errno= sql_errno_arg;
379
strmake(m_message, message_arg, sizeof(m_message) - 1);
386
Mark the diagnostics area as 'DISABLED'.
388
This is used in rare cases when the COM_ command at hand sends a response
389
in a custom format. One example is the query cache, another is
394
Diagnostics_area::disable_status()
397
m_status= DA_DISABLED;
117
int session_in_lock_tables(const Session *session)
119
return test(session->in_lock_tables);
124
int session_tablespace_op(const Session *session)
126
return test(session->tablespace_op);
131
Set the process info field of the Session structure.
133
This function is used by plug-ins. Internally, the
134
Session::set_proc_info() function should be used.
136
@see Session::set_proc_info
139
set_session_proc_info(Session *session, const char *info)
141
session->set_proc_info(info);
145
const char *get_session_proc_info(Session *session)
147
return session->get_proc_info();
151
void **session_ha_data(const Session *session, const struct StorageEngine *engine)
153
return (void **) &session->ha_data[engine->slot].ha_ptr;
157
int64_t session_test_options(const Session *session, int64_t test_options)
159
return session->options & test_options;
163
int session_sql_command(const Session *session)
165
return (int) session->lex->sql_command;
169
int session_tx_isolation(const Session *session)
171
return (int) session->variables.tx_isolation;
175
void session_inc_row_count(Session *session)
177
session->row_count++;
180
Session::Session(Protocol *protocol_arg)
402
181
:Statement(&main_lex, &main_mem_root,
403
182
/* statement id */ 0),
404
Open_tables_state(refresh_version), rli_fake(0),
183
Open_tables_state(refresh_version),
405
184
lock_id(&main_lock_id),
406
user_time(0), in_sub_stmt(0),
407
binlog_table_maps(0), binlog_flags(0UL),
408
186
arg_of_last_insert_id_function(false),
409
187
first_successful_insert_id_in_prev_stmt(0),
410
first_successful_insert_id_in_prev_stmt_for_binlog(0),
411
188
first_successful_insert_id_in_cur_stmt(0),
412
stmt_depends_on_first_successful_insert_id_in_prev_stmt(false),
413
189
global_read_lock(0),
414
190
is_fatal_error(0),
415
191
transaction_rollback_request(0),
416
192
is_fatal_sub_stmt_error(0),
419
193
in_lock_tables(0),
421
194
derived_tables_processing(false),
200
memset(process_list_info, 0, PROCESS_LIST_WIDTH);
427
203
Pass nominal parameters to init_alloc_root only to ensure that
841
573
assert(thread_stack);
843
if (my_pthread_setspecific_ptr(THR_THD, this) ||
844
my_pthread_setspecific_ptr(THR_MALLOC, &mem_root))
575
if (pthread_setspecific(THR_Session, this) ||
576
pthread_setspecific(THR_Mem_root, &mem_root))
846
578
mysys_var=my_thread_var;
848
581
Let mysqld define the thread id (not mysys)
849
This allows us to move THD to different threads if needed.
582
This allows us to move Session to different threads if needed.
851
584
mysys_var->id= thread_id;
852
585
real_id= pthread_self(); // For debugging
855
We have to call thr_lock_info_init() again here as THD may have been
588
We have to call thr_lock_info_init() again here as Session may have been
856
589
created in another thread
858
591
thr_lock_info_init(&lock_info);
595
void Session::prepareForQueries()
597
if (variables.max_join_size == HA_POS_ERROR)
598
options |= OPTION_BIG_SELECTS;
599
if (client_capabilities & CLIENT_COMPRESS)
602
protocol->enableCompression();
605
version= refresh_version;
612
bool Session::initGlobals()
616
disconnect(ER_OUT_OF_RESOURCES, true);
617
statistic_increment(aborted_connects, &LOCK_status);
618
Scheduler &thread_scheduler= get_thread_scheduler();
619
thread_scheduler.end_thread(this, 0);
625
bool Session::authenticate()
628
if (protocol->authenticate())
631
statistic_increment(aborted_connects, &LOCK_status);
635
bool Session::checkUser(const char *passwd, uint32_t passwd_len, const char *in_db)
637
LEX_STRING db_str= { (char *) in_db, in_db ? strlen(in_db) : 0 };
638
bool is_authenticated;
641
Clear session->db as it points to something, that will be freed when
642
connection is closed. We don't want to accidentally free a wrong
643
pointer if connect failed. Also in case of 'CHANGE USER' failure,
644
current database will be switched to 'no database selected'.
648
if (passwd_len != 0 && passwd_len != SCRAMBLE_LENGTH)
650
my_error(ER_HANDSHAKE_ERROR, MYF(0), security_ctx.ip.c_str());
654
is_authenticated= authenticate_user(this, passwd);
656
if (is_authenticated != true)
658
my_error(ER_ACCESS_DENIED_ERROR, MYF(0),
659
security_ctx.user.c_str(),
660
security_ctx.ip.c_str(),
661
passwd_len ? ER(ER_YES) : ER(ER_NO));
666
security_ctx.skip_grants();
668
/* Change database if necessary */
669
if (in_db && in_db[0])
671
if (mysql_change_db(this, &db_str, false))
673
/* mysql_change_db() has pushed the error message. */
678
password= test(passwd_len); // remember for error messages
680
/* Ready to handle queries */
684
bool Session::executeStatement()
687
uint32_t packet_length;
689
enum enum_server_command l_command;
692
indicator of uninitialized lex => normal flow of errors handling
695
lex->current_select= 0;
697
if (protocol->readCommand(&l_packet, &packet_length) == false)
700
if (packet_length == 0)
703
l_command= (enum enum_server_command) (unsigned char) l_packet[0];
705
if (command >= COM_END)
706
command= COM_END; // Wrong command
708
assert(packet_length);
709
return ! dispatch_command(l_command, this, l_packet+1, (uint32_t) (packet_length-1));
712
bool Session::readAndStoreQuery(const char *in_packet, uint32_t in_packet_length)
714
/* Remove garbage at start and end of query */
715
while (in_packet_length > 0 && my_isspace(charset(), in_packet[0]))
720
const char *pos= in_packet + in_packet_length; /* Point at end null */
721
while (in_packet_length > 0 &&
722
(pos[-1] == ';' || my_isspace(charset() ,pos[-1])))
728
/* We must allocate some extra memory for the cached query string */
729
query_length= 0; /* Extra safety: Avoid races */
730
query= (char*) memdup_w_gap((unsigned char*) in_packet, in_packet_length, db_length + 1);
734
query[in_packet_length]=0;
735
query_length= in_packet_length;
737
/* Reclaim some memory */
738
packet.shrink(variables.net_buffer_length);
739
convert_buffer.shrink(variables.net_buffer_length);
744
bool Session::endTransaction(enum enum_mysql_completiontype completion)
749
if (transaction.xid_state.xa_state != XA_NOTR)
751
my_error(ER_XAER_RMFAIL, MYF(0), xa_state_names[transaction.xid_state.xa_state]);
758
* We don't use endActiveTransaction() here to ensure that this works
759
* even if there is a problem with the OPTION_AUTO_COMMIT flag
760
* (Which of course should never happen...)
762
server_status&= ~SERVER_STATUS_IN_TRANS;
765
options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
766
transaction.all.modified_non_trans_table= false;
769
do_release= 1; /* fall through */
770
case COMMIT_AND_CHAIN:
771
result= endActiveTransaction();
772
if (result == true && completion == COMMIT_AND_CHAIN)
773
result= startTransaction();
775
case ROLLBACK_RELEASE:
776
do_release= 1; /* fall through */
778
case ROLLBACK_AND_CHAIN:
780
server_status&= ~SERVER_STATUS_IN_TRANS;
781
if (ha_rollback(this))
783
options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
784
transaction.all.modified_non_trans_table= false;
785
if (result == true && (completion == ROLLBACK_AND_CHAIN))
786
result= startTransaction();
790
my_error(ER_UNKNOWN_COM_ERROR, MYF(0));
795
my_error(killed_errno(), MYF(0));
796
else if ((result == true) && do_release)
797
killed= Session::KILL_CONNECTION;
802
bool Session::endActiveTransaction()
806
if (transaction.xid_state.xa_state != XA_NOTR)
808
my_error(ER_XAER_RMFAIL, MYF(0), xa_state_names[transaction.xid_state.xa_state]);
811
if (options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN | OPTION_TABLE_LOCK))
813
/* Safety if one did "drop table" on locked tables */
815
options&= ~OPTION_TABLE_LOCK;
816
server_status&= ~SERVER_STATUS_IN_TRANS;
820
options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
821
transaction.all.modified_non_trans_table= false;
825
bool Session::startTransaction()
832
locked_tables= 0; // Will be automatically closed
833
close_thread_tables(this); // Free tables
835
if (! endActiveTransaction())
839
options|= OPTION_BEGIN;
840
server_status|= SERVER_STATUS_IN_TRANS;
841
if (lex->start_transaction_opt & DRIZZLE_START_TRANS_OPT_WITH_CONS_SNAPSHOT)
842
if (ha_start_consistent_snapshot(this))
864
849
Cleanup after query.
867
THD::cleanup_after_query()
852
Session::cleanup_after_query()
870
855
This function is used to reset thread data to its default state.
1141
1116
Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF));
1145
struct Item_change_record: public ilink
1149
/* Placement new was hidden by `new' in ilink (TODO: check): */
1150
static void *operator new(size_t size __attribute__((unused)),
1153
static void operator delete(void *ptr __attribute__((unused)),
1154
size_t size __attribute__((unused)))
1156
static void operator delete(void *ptr __attribute__((unused)),
1157
void *mem __attribute__((unused)))
1158
{ /* never called */ }
1163
Register an item tree tree transformation, performed by the query
1164
optimizer. We need a pointer to runtime_memroot because it may be !=
1165
thd->mem_root (this may no longer be a true statement)
1168
void THD::nocheck_register_item_tree_change(Item **place, Item *old_value,
1169
MEM_ROOT *runtime_memroot)
1171
Item_change_record *change;
1173
Now we use one node per change, which adds some memory overhead,
1174
but still is rather fast as we use alloc_root for allocations.
1175
A list of item tree changes of an average query should be short.
1177
void *change_mem= alloc_root(runtime_memroot, sizeof(*change));
1178
if (change_mem == 0)
1181
OOM, thd->fatal_error() is called by the error handler of the
1182
memroot. Just return.
1186
change= new (change_mem) Item_change_record;
1187
change->place= place;
1188
change->old_value= old_value;
1189
change_list.append(change);
1193
void THD::rollback_item_tree_changes()
1195
I_List_iterator<Item_change_record> it(change_list);
1196
Item_change_record *change;
1198
while ((change= it++))
1199
*change->place= change->old_value;
1200
/* We can forget about changes memory: it's allocated in runtime memroot */
1201
change_list.empty();
1206
/*****************************************************************************
1207
** Functions to provide a interface to select results
1208
*****************************************************************************/
1210
select_result::select_result()
1215
void select_result::send_error(uint32_t errcode,const char *err)
1217
my_message(errcode, err, MYF(0));
1221
void select_result::cleanup()
1226
bool select_result::check_simple_select() const
1228
my_error(ER_SP_BAD_CURSOR_QUERY, MYF(0));
1233
static String default_line_term("\n",default_charset_info);
1234
static String default_escaped("\\",default_charset_info);
1235
static String default_field_term("\t",default_charset_info);
1237
sql_exchange::sql_exchange(char *name, bool flag,
1238
enum enum_filetype filetype_arg)
1239
:file_name(name), opt_enclosed(0), dumpfile(flag), skip_lines(0)
1241
filetype= filetype_arg;
1242
field_term= &default_field_term;
1243
enclosed= line_start= &my_empty_string;
1244
line_term= &default_line_term;
1245
escaped= &default_escaped;
1249
bool select_send::send_fields(List<Item> &list, uint32_t flags)
1252
if (!(res= thd->protocol->send_fields(&list, flags)))
1253
is_result_set_started= 1;
1257
void select_send::abort()
1264
Cleanup an instance of this class for re-use
1265
at next execution of a prepared statement/
1266
stored procedure statement.
1269
void select_send::cleanup()
1271
is_result_set_started= false;
1274
/* Send data to client. Returns 0 if ok */
1276
bool select_send::send_data(List<Item> &items)
1278
if (unit->offset_limit_cnt)
1279
{ // using limit offset,count
1280
unit->offset_limit_cnt--;
1285
We may be passing the control from mysqld to the client: release the
1286
InnoDB adaptive hash S-latch to avoid thread deadlocks if it was reserved
1289
ha_release_temporary_latches(thd);
1291
List_iterator_fast<Item> li(items);
1292
Protocol *protocol= thd->protocol;
1293
char buff[MAX_FIELD_WIDTH];
1294
String buffer(buff, sizeof(buff), &my_charset_bin);
1296
protocol->prepare_for_resend();
1300
if (item->send(protocol, &buffer))
1302
protocol->free(); // Free used buffer
1303
my_message(ER_OUT_OF_RESOURCES, ER(ER_OUT_OF_RESOURCES), MYF(0));
1307
thd->sent_row_count++;
1308
if (thd->is_error())
1310
protocol->remove_last_row();
1314
return(protocol->write());
1318
bool select_send::send_eof()
1321
We may be passing the control from mysqld to the client: release the
1322
InnoDB adaptive hash S-latch to avoid thread deadlocks if it was reserved
1325
ha_release_temporary_latches(thd);
1327
/* Unlock tables before sending packet to gain some speed */
1330
mysql_unlock_tables(thd, thd->lock);
1334
is_result_set_started= 0;
1339
1119
/************************************************************************
1340
1120
Handling writing to file
1341
1121
************************************************************************/
1850
bool Session::set_db(const char *new_db, size_t new_db_len)
1852
/* Do not reallocate memory if current chunk is big enough. */
1853
if (db && new_db && db_length >= new_db_len)
1854
memcpy(db, new_db, new_db_len+1);
1861
db= (char *)malloc(new_db_len + 1);
1864
memcpy(db, new_db, new_db_len);
1871
db_length= db ? new_db_len : 0;
1872
return new_db && !db;
2175
1877
Check the killed state of a user thread
2176
@param thd user thread
1878
@param session user thread
2177
1879
@retval 0 the user thread is active
2178
1880
@retval 1 the user thread has been killed
2180
extern "C" int thd_killed(const DRIZZLE_THD thd)
1882
extern "C" int session_killed(const Session *session)
2182
return(thd->killed);
1884
return(session->killed);
2186
1888
Return the thread id of a user thread
2187
@param thd user thread
1889
@param session user thread
2188
1890
@return thread id
2190
extern "C" unsigned long thd_get_thread_id(const DRIZZLE_THD thd)
2192
return((unsigned long)thd->thread_id);
2196
#ifdef INNODB_COMPATIBILITY_HOOKS
2197
extern "C" const struct charset_info_st *thd_charset(DRIZZLE_THD thd)
2199
return(thd->charset());
2202
extern "C" char **thd_query(DRIZZLE_THD thd)
2204
return(&thd->query);
2207
extern "C" int thd_slave_thread(const DRIZZLE_THD thd)
2209
return(thd->slave_thread);
2212
extern "C" int thd_non_transactional_update(const DRIZZLE_THD thd)
2214
return(thd->transaction.all.modified_non_trans_table);
2217
extern "C" int thd_binlog_format(const DRIZZLE_THD thd)
2219
return (int) thd->variables.binlog_format;
2222
extern "C" void thd_mark_transaction_to_rollback(DRIZZLE_THD thd, bool all)
2224
mark_transaction_to_rollback(thd, all);
2226
#endif // INNODB_COMPATIBILITY_HOOKS */
1892
extern "C" unsigned long session_get_thread_id(const Session *session)
1894
return((unsigned long)session->thread_id);
1899
LEX_STRING *session_make_lex_string(Session *session, LEX_STRING *lex_str,
1900
const char *str, unsigned int size,
1901
int allocate_lex_string)
1903
return session->make_lex_string(lex_str, str, size,
1904
(bool) allocate_lex_string);
1907
extern "C" const struct charset_info_st *session_charset(Session *session)
1909
return(session->charset());
1912
extern "C" char **session_query(Session *session)
1914
return(&session->query);
1917
extern "C" int session_non_transactional_update(const Session *session)
1919
return(session->transaction.all.modified_non_trans_table);
1922
extern "C" void session_mark_transaction_to_rollback(Session *session, bool all)
1924
mark_transaction_to_rollback(session, all);
2230
1929
Mark transaction to rollback and mark error as fatal to a sub-statement.
2232
@param thd Thread handle
1931
@param session Thread handle
2233
1932
@param all true <=> rollback main transaction.
2236
void mark_transaction_to_rollback(THD *thd, bool all)
2240
thd->is_fatal_sub_stmt_error= true;
2241
thd->transaction_rollback_request= all;
2244
/***************************************************************************
2245
Handling of XA id cacheing
2246
***************************************************************************/
2248
pthread_mutex_t LOCK_xid_cache;
2251
extern "C" unsigned char *xid_get_hash_key(const unsigned char *, size_t *, bool);
2252
extern "C" void xid_free_hash(void *);
2254
unsigned char *xid_get_hash_key(const unsigned char *ptr, size_t *length,
2255
bool not_used __attribute__((unused)))
2257
*length=((XID_STATE*)ptr)->xid.key_length();
2258
return ((XID_STATE*)ptr)->xid.key();
2261
void xid_free_hash(void *ptr)
2263
if (!((XID_STATE*)ptr)->in_thd)
2264
free((unsigned char*)ptr);
2267
bool xid_cache_init()
2269
pthread_mutex_init(&LOCK_xid_cache, MY_MUTEX_INIT_FAST);
2270
return hash_init(&xid_cache, &my_charset_bin, 100, 0, 0,
2271
xid_get_hash_key, xid_free_hash, 0) != 0;
2274
void xid_cache_free()
2276
if (hash_inited(&xid_cache))
2278
hash_free(&xid_cache);
2279
pthread_mutex_destroy(&LOCK_xid_cache);
2283
XID_STATE *xid_cache_search(XID *xid)
2285
pthread_mutex_lock(&LOCK_xid_cache);
2286
XID_STATE *res=(XID_STATE *)hash_search(&xid_cache, xid->key(), xid->key_length());
2287
pthread_mutex_unlock(&LOCK_xid_cache);
2292
bool xid_cache_insert(XID *xid, enum xa_states xa_state)
2296
pthread_mutex_lock(&LOCK_xid_cache);
2297
if (hash_search(&xid_cache, xid->key(), xid->key_length()))
2299
else if (!(xs=(XID_STATE *)my_malloc(sizeof(*xs), MYF(MY_WME))))
2303
xs->xa_state=xa_state;
2306
res=my_hash_insert(&xid_cache, (unsigned char*)xs);
2308
pthread_mutex_unlock(&LOCK_xid_cache);
2313
bool xid_cache_insert(XID_STATE *xid_state)
2315
pthread_mutex_lock(&LOCK_xid_cache);
2316
assert(hash_search(&xid_cache, xid_state->xid.key(),
2317
xid_state->xid.key_length())==0);
2318
bool res=my_hash_insert(&xid_cache, (unsigned char*)xid_state);
2319
pthread_mutex_unlock(&LOCK_xid_cache);
2324
void xid_cache_delete(XID_STATE *xid_state)
2326
pthread_mutex_lock(&LOCK_xid_cache);
2327
hash_delete(&xid_cache, (unsigned char *)xid_state);
2328
pthread_mutex_unlock(&LOCK_xid_cache);
2332
Implementation of interface to write rows to the binary log through the
2333
thread. The thread is responsible for writing the rows it has
2334
inserted/updated/deleted.
2339
Template member function for ensuring that there is an rows log
2340
event of the apropriate type before proceeding.
2343
- Events of type 'RowEventT' have the type code 'type_code'.
2346
If a non-NULL pointer is returned, the pending event for thread 'thd' will
2347
be an event of type 'RowEventT' (which have the type code 'type_code')
2348
will either empty or have enough space to hold 'needed' bytes. In
2349
addition, the columns bitmap will be correct for the row, meaning that
2350
the pending event will be flushed if the columns in the event differ from
2351
the columns suppled to the function.
2354
If no error, a non-NULL pending event (either one which already existed or
2355
the newly created one).
2359
template <class RowsEventT> Rows_log_event*
2360
THD::binlog_prepare_pending_rows_event(Table* table, uint32_t serv_id,
2362
bool is_transactional,
2363
RowsEventT *hint __attribute__((unused)))
2365
/* Pre-conditions */
2366
assert(table->s->table_map_id != UINT32_MAX);
2368
/* Fetch the type code for the RowsEventT template parameter */
2369
int const type_code= RowsEventT::TYPE_CODE;
2372
There is no good place to set up the transactional data, so we
2375
if (binlog_setup_trx_data())
2378
Rows_log_event* pending= binlog_get_pending_rows_event();
2380
if (unlikely(pending && !pending->is_valid()))
2384
Check if the current event is non-NULL and a write-rows
2385
event. Also check if the table provided is mapped: if it is not,
2386
then we have switched to writing to a new table.
2387
If there is no pending event, we need to create one. If there is a pending
2388
event, but it's not about the same table id, or not of the same type
2389
(between Write, Update and Delete), or not the same affected columns, or
2390
going to be too big, flush this event to disk and create a new pending
2393
The last test is necessary for the Cluster injector to work
2394
correctly. The reason is that the Cluster can inject two write
2395
rows with different column bitmaps if there is an insert followed
2396
by an update in the same transaction, and these are grouped into a
2397
single epoch/transaction when fed to the injector.
2399
TODO: Fix the code so that the last test can be removed.
2402
pending->server_id != serv_id ||
2403
pending->get_table_id() != table->s->table_map_id ||
2404
pending->get_type_code() != type_code ||
2405
pending->get_data_size() + needed > opt_binlog_rows_event_max_size ||
2406
!bitmap_cmp(pending->get_cols(), table->write_set))
2408
/* Create a new RowsEventT... */
2409
Rows_log_event* const
2410
ev= new RowsEventT(this, table, table->s->table_map_id,
2414
ev->server_id= serv_id; // I don't like this, it's too easy to forget.
2416
flush the pending event and replace it with the newly created
2419
if (unlikely(mysql_bin_log.flush_and_set_pending_rows_event(this, ev)))
2425
return(ev); /* This is the new pending event */
2427
return(pending); /* This is the current pending event */
2430
#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
2432
Instantiate the versions we need, we have -fno-implicit-template as
2435
template Rows_log_event*
2436
THD::binlog_prepare_pending_rows_event(Table*, uint32_t, size_t, bool,
2437
Write_rows_log_event*);
2439
template Rows_log_event*
2440
THD::binlog_prepare_pending_rows_event(Table*, uint32_t, size_t, bool,
2441
Delete_rows_log_event *);
2443
template Rows_log_event*
2444
THD::binlog_prepare_pending_rows_event(Table*, uint32_t, size_t, bool,
2445
Update_rows_log_event *);
2450
Class to handle temporary allocation of memory for row data.
2452
The responsibilities of the class is to provide memory for
2453
packing one or two rows of packed data (depending on what
2454
constructor is called).
2456
In order to make the allocation more efficient for "simple" rows,
2457
i.e., rows that do not contain any blobs, a pointer to the
2458
allocated memory is of memory is stored in the table structure
2459
for simple rows. If memory for a table containing a blob field
2460
is requested, only memory for that is allocated, and subsequently
2461
released when the object is destroyed.
2464
class Row_data_memory {
2467
Build an object to keep track of a block-local piece of memory
2468
for storing a row of data.
2471
Table where the pre-allocated memory is stored.
2474
Length of data that is needed, if the record contain blobs.
2476
Row_data_memory(Table *table, size_t const len1)
2479
m_alloc_checked= false;
2480
allocate_memory(table, len1);
2481
m_ptr[0]= has_memory() ? m_memory : 0;
2485
Row_data_memory(Table *table, size_t const len1, size_t const len2)
2488
m_alloc_checked= false;
2489
allocate_memory(table, len1 + len2);
2490
m_ptr[0]= has_memory() ? m_memory : 0;
2491
m_ptr[1]= has_memory() ? m_memory + len1 : 0;
2496
if (m_memory != 0 && m_release_memory_on_destruction)
2497
free((unsigned char*) m_memory);
2501
Is there memory allocated?
2503
@retval true There is memory allocated
2504
@retval false Memory allocation failed
2506
bool has_memory() const {
2507
m_alloc_checked= true;
2508
return m_memory != 0;
2511
unsigned char *slot(uint32_t s)
2513
assert(s < sizeof(m_ptr)/sizeof(*m_ptr));
2514
assert(m_ptr[s] != 0);
2515
assert(m_alloc_checked == true);
2520
void allocate_memory(Table *const table, size_t const total_length)
2522
if (table->s->blob_fields == 0)
2525
The maximum length of a packed record is less than this
2526
length. We use this value instead of the supplied length
2527
when allocating memory for records, since we don't know how
2528
the memory will be used in future allocations.
2530
Since table->s->reclength is for unpacked records, we have
2531
to add two bytes for each field, which can potentially be
2532
added to hold the length of a packed field.
2534
size_t const maxlen= table->s->reclength + 2 * table->s->fields;
2537
Allocate memory for two records if memory hasn't been
2538
allocated. We allocate memory for two records so that it can
2539
be used when processing update rows as well.
2541
if (table->write_row_record == 0)
2542
table->write_row_record=
2543
(unsigned char *) alloc_root(&table->mem_root, 2 * maxlen);
2544
m_memory= table->write_row_record;
2545
m_release_memory_on_destruction= false;
2549
m_memory= (unsigned char *) my_malloc(total_length, MYF(MY_WME));
2550
m_release_memory_on_destruction= true;
2554
mutable bool m_alloc_checked;
2555
bool m_release_memory_on_destruction;
2556
unsigned char *m_memory;
2557
unsigned char *m_ptr[2];
2562
int THD::binlog_write_row(Table* table, bool is_trans,
2563
unsigned char const *record)
2565
assert(current_stmt_binlog_row_based && mysql_bin_log.is_open());
2568
Pack records into format for transfer. We are allocating more
2569
memory than needed, but that doesn't matter.
2571
Row_data_memory memory(table, table->max_row_length(record));
2572
if (!memory.has_memory())
2573
return HA_ERR_OUT_OF_MEM;
2575
unsigned char *row_data= memory.slot(0);
2577
size_t const len= pack_row(table, table->write_set, row_data, record);
2579
Rows_log_event* const ev=
2580
binlog_prepare_pending_rows_event(table, server_id, len, is_trans,
2581
static_cast<Write_rows_log_event*>(0));
2583
if (unlikely(ev == 0))
2584
return HA_ERR_OUT_OF_MEM;
2586
return ev->add_row_data(row_data, len);
2589
int THD::binlog_update_row(Table* table, bool is_trans,
2590
const unsigned char *before_record,
2591
const unsigned char *after_record)
2593
assert(current_stmt_binlog_row_based && mysql_bin_log.is_open());
2595
size_t const before_maxlen = table->max_row_length(before_record);
2596
size_t const after_maxlen = table->max_row_length(after_record);
2598
Row_data_memory row_data(table, before_maxlen, after_maxlen);
2599
if (!row_data.has_memory())
2600
return HA_ERR_OUT_OF_MEM;
2602
unsigned char *before_row= row_data.slot(0);
2603
unsigned char *after_row= row_data.slot(1);
2605
size_t const before_size= pack_row(table, table->read_set, before_row,
2607
size_t const after_size= pack_row(table, table->write_set, after_row,
2610
Rows_log_event* const ev=
2611
binlog_prepare_pending_rows_event(table, server_id,
2612
before_size + after_size, is_trans,
2613
static_cast<Update_rows_log_event*>(0));
2615
if (unlikely(ev == 0))
2616
return HA_ERR_OUT_OF_MEM;
2619
ev->add_row_data(before_row, before_size) ||
2620
ev->add_row_data(after_row, after_size);
2623
int THD::binlog_delete_row(Table* table, bool is_trans,
2624
unsigned char const *record)
2626
assert(current_stmt_binlog_row_based && mysql_bin_log.is_open());
2629
Pack records into format for transfer. We are allocating more
2630
memory than needed, but that doesn't matter.
2632
Row_data_memory memory(table, table->max_row_length(record));
2633
if (unlikely(!memory.has_memory()))
2634
return HA_ERR_OUT_OF_MEM;
2636
unsigned char *row_data= memory.slot(0);
2638
size_t const len= pack_row(table, table->read_set, row_data, record);
2640
Rows_log_event* const ev=
2641
binlog_prepare_pending_rows_event(table, server_id, len, is_trans,
2642
static_cast<Delete_rows_log_event*>(0));
2644
if (unlikely(ev == 0))
2645
return HA_ERR_OUT_OF_MEM;
2647
return ev->add_row_data(row_data, len);
2651
int THD::binlog_flush_pending_rows_event(bool stmt_end)
2654
We shall flush the pending event even if we are not in row-based
2655
mode: it might be the case that we left row-based mode before
2656
flushing anything (e.g., if we have explicitly locked tables).
2658
if (!mysql_bin_log.is_open())
2662
Mark the event as the last event of a statement if the stmt_end
2666
if (Rows_log_event *pending= binlog_get_pending_rows_event())
2670
pending->set_flags(Rows_log_event::STMT_END_F);
2671
pending->flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
2672
binlog_table_maps= 0;
2675
error= mysql_bin_log.flush_and_set_pending_rows_event(this, 0);
2683
Member function that will log query, either row-based or
2684
statement-based depending on the value of the 'current_stmt_binlog_row_based'
2685
the value of the 'qtype' flag.
2687
This function should be called after the all calls to ha_*_row()
2688
functions have been issued, but before tables are unlocked and
2692
There shall be no writes to any system table after calling
2693
binlog_query(), so these writes has to be moved to before the call
2694
of binlog_query() for correct functioning.
2696
This is necessesary not only for RBR, but the master might crash
2697
after binlogging the query but before changing the system tables.
2698
This means that the slave and the master are not in the same state
2699
(after the master has restarted), so therefore we have to
2700
eliminate this problem.
2703
Error code, or 0 if no error.
2705
int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
2706
ulong query_len, bool is_trans, bool suppress_use,
2707
THD::killed_state killed_status_arg)
2709
assert(query_arg && mysql_bin_log.is_open());
2711
if (int error= binlog_flush_pending_rows_event(true))
2715
If we are in statement mode and trying to log an unsafe statement,
2716
we should print a warning.
2718
if (lex->is_stmt_unsafe() &&
2719
variables.binlog_format == BINLOG_FORMAT_STMT)
2721
assert(this->query != NULL);
2722
push_warning(this, DRIZZLE_ERROR::WARN_LEVEL_WARN,
2723
ER_BINLOG_UNSAFE_STATEMENT,
2724
ER(ER_BINLOG_UNSAFE_STATEMENT));
2725
if (!(binlog_flags & BINLOG_FLAG_UNSAFE_STMT_PRINTED))
2727
char warn_buf[DRIZZLE_ERRMSG_SIZE];
2728
snprintf(warn_buf, DRIZZLE_ERRMSG_SIZE, "%s Statement: %s",
2729
ER(ER_BINLOG_UNSAFE_STATEMENT), this->query);
2730
sql_print_warning(warn_buf);
2731
binlog_flags|= BINLOG_FLAG_UNSAFE_STMT_PRINTED;
2736
case THD::ROW_QUERY_TYPE:
2737
if (current_stmt_binlog_row_based)
2739
/* Otherwise, we fall through */
2740
case THD::DRIZZLE_QUERY_TYPE:
2742
Using this query type is a conveniece hack, since we have been
2743
moving back and forth between using RBR for replication of
2744
system tables and not using it.
2746
Make sure to change in check_table_binlog_row_based() according
2747
to how you treat this.
2749
case THD::STMT_QUERY_TYPE:
2751
The DRIZZLE_LOG::write() function will set the STMT_END_F flag and
2752
flush the pending rows event if necessary.
2755
Query_log_event qinfo(this, query_arg, query_len, is_trans, suppress_use,
2757
qinfo.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
2759
Binlog table maps will be irrelevant after a Query_log_event
2760
(they are just removed on the slave side) so after the query
2761
log event is written to the binary log, we pretend that no
2762
table maps were written.
2764
int error= mysql_bin_log.write(&qinfo);
2765
binlog_table_maps= 0;
2770
case THD::QUERY_TYPE_COUNT:
2772
assert(0 <= qtype && qtype < QUERY_TYPE_COUNT);
2777
bool Discrete_intervals_list::append(uint64_t start, uint64_t val,
2780
/* first, see if this can be merged with previous */
2781
if ((head == NULL) || tail->merge_if_contiguous(start, val, incr))
2783
/* it cannot, so need to add a new interval */
2784
Discrete_interval *new_interval= new Discrete_interval(start, val, incr);
2785
return(append(new_interval));
2790
bool Discrete_intervals_list::append(Discrete_interval *new_interval)
2792
if (unlikely(new_interval == NULL))
2795
head= current= new_interval;
2797
tail->next= new_interval;
1935
void mark_transaction_to_rollback(Session *session, bool all)
1939
session->is_fatal_sub_stmt_error= true;
1940
session->transaction_rollback_request= all;
1944
void Session::disconnect(uint32_t errcode, bool should_lock)
1946
/* Allow any plugins to cleanup their session variables */
1947
plugin_sessionvar_cleanup(this);
1949
/* If necessary, log any aborted or unauthorized connections */
1950
if (killed || protocol->wasAborted())
1951
statistic_increment(aborted_threads, &LOCK_status);
1953
if (protocol->wasAborted())
1955
if (! killed && variables.log_warnings > 1)
1957
Security_context *sctx= &security_ctx;
1959
errmsg_printf(ERRMSG_LVL_WARN, ER(ER_NEW_ABORTING_CONNECTION)
1961
, (db ? db : "unconnected")
1962
, sctx->user.empty() == false ? sctx->user.c_str() : "unauthenticated"
1964
, (main_da.is_error() ? main_da.message() : ER(ER_UNKNOWN_ERROR)));
1968
/* Close out our connection to the client */
1970
(void) pthread_mutex_lock(&LOCK_thread_count);
1971
killed= Session::KILL_CONNECTION;
1972
if (protocol->isConnected())
1976
/*my_error(errcode, ER(errcode));*/
1977
protocol->sendError(errcode, ER(errcode)); /* purecov: inspected */
1982
(void) pthread_mutex_unlock(&LOCK_thread_count);
1986
Reset Session part responsible for command processing state.
1988
This needs to be called before execution of every statement
1989
(prepared or conventional).
1990
It is not called by substatements of routines.
1993
Make it a method of Session and align its name with the rest of
1994
reset/end/start/init methods.
1996
Call it after we use Session for queries, not before.
1999
void Session::reset_for_next_command()
2004
Those two lines below are theoretically unneeded as
2005
Session::cleanup_after_query() should take care of this already.
2007
auto_inc_intervals_in_cur_stmt_for_binlog.empty();
2010
server_status&= ~ (SERVER_MORE_RESULTS_EXISTS |
2011
SERVER_QUERY_NO_INDEX_USED |
2012
SERVER_QUERY_NO_GOOD_INDEX_USED);
2014
If in autocommit mode and not in a transaction, reset
2015
OPTION_STATUS_NO_TRANS_UPDATE | OPTION_KEEP_LOG to not get warnings
2016
in ha_rollback_trans() about some tables couldn't be rolled back.
2018
if (!(options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
2020
options&= ~OPTION_KEEP_LOG;
2021
transaction.all.modified_non_trans_table= false;
2025
main_da.reset_diagnostics_area();
2026
total_warn_count=0; // Warnings for this query
2027
sent_row_count= examined_row_count= 0;
2033
Close all temporary tables created by 'CREATE TEMPORARY TABLE' for thread
2034
creates one DROP TEMPORARY Table binlog event for each pseudo-thread
2037
void Session::close_temporary_tables()
2042
if (!temporary_tables)
2045
for (table= temporary_tables; table; table= tmp_next)
2047
tmp_next= table->next;
2048
close_temporary(table, 1, 1);
2050
temporary_tables= 0;
2056
/** Clear most status variables. */
2057
extern time_t flush_status_time;
2058
extern uint32_t max_used_connections;
2060
void Session::refresh_status()
2062
pthread_mutex_lock(&LOCK_status);
2064
/* Add thread's status variabes to global status */
2065
add_to_status(&global_status_var, &status_var);
2067
/* Reset thread's status variables */
2068
memset(&status_var, 0, sizeof(status_var));
2070
/* Reset some global variables */
2071
reset_status_vars();
2073
/* Reset the counters of all key caches (default and named). */
2074
process_key_caches(reset_key_cache_counters);
2075
flush_status_time= time((time_t*) 0);
2076
max_used_connections= 1; /* We set it to one, because we know we exist */
2077
pthread_mutex_unlock(&LOCK_status);
2080
#define extra_size sizeof(double)
2082
user_var_entry *Session::getVariable(LEX_STRING &name, bool create_if_not_exists)
2084
user_var_entry *entry= NULL;
2086
assert(name.length == strlen (name.str));
2087
entry= (user_var_entry*) hash_search(&user_vars, (unsigned char*) name.str, name.length);
2089
if ((entry == NULL) && create_if_not_exists)
2091
uint32_t size=ALIGN_SIZE(sizeof(user_var_entry))+name.length+1+extra_size;
2092
if (!hash_inited(&user_vars))
2094
if (!(entry = (user_var_entry*) malloc(size)))
2096
entry->name.str=(char*) entry+ ALIGN_SIZE(sizeof(user_var_entry))+
2098
entry->name.length=name.length;
2101
entry->update_query_id=0;
2102
entry->collation.set(NULL, DERIVATION_IMPLICIT, 0);
2103
entry->unsigned_flag= 0;
2105
If we are here, we were called from a SET or a query which sets a
2106
variable. Imagine it is this:
2107
INSERT INTO t SELECT @a:=10, @a:=@a+1.
2108
Then when we have a Item_func_get_user_var (because of the @a+1) so we
2109
think we have to write the value of @a to the binlog. But before that,
2110
we have a Item_func_set_user_var to create @a (@a:=10), in this we mark
2111
the variable as "already logged" (line below) so that it won't be logged
2112
by Item_func_get_user_var (because that's not necessary).
2114
entry->used_query_id= query_id;
2115
entry->type=STRING_RESULT;
2116
memcpy(entry->name.str, name.str, name.length+1);
2117
if (my_hash_insert(&user_vars, (unsigned char*) entry))
2120
free((char*) entry);