← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_update.cc

  • Committer: Jay Pipes
  • Date: 2009-02-28 20:43:31 UTC
  • mto: (910.2.6 mordred-noatomics)
  • mto: This revision was merged to the branch mainline in revision 912.
  • Revision ID: jpipes@serialcoder-20090228204331-6x804cdbfzyy9w8i
Merged in remove-timezone work

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_update.cc
11
11
 
12
12
   You should have received a copy of the GNU General Public License
13
13
   along with this program; if not, write to the Free Software
14
 
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
15
15
 
16
16
 
17
17
/*
18
18
  Single table and multi table updates of tables.
19
19
  Multi-table updates were introduced by Sinisa & Monty
20
20
*/
21
 
#include "config.h"
22
 
#include "drizzled/sql_select.h"
23
 
#include "drizzled/error.h"
24
 
#include "drizzled/probes.h"
25
 
#include "drizzled/sql_base.h"
26
 
#include "drizzled/field/timestamp.h"
27
 
#include "drizzled/sql_parse.h"
28
 
#include "drizzled/optimizer/range.h"
29
 
#include "drizzled/records.h"
30
 
#include "drizzled/internal/my_sys.h"
31
 
#include "drizzled/internal/iocache.h"
32
 
#include "drizzled/transaction_services.h"
33
 
#include "drizzled/filesort.h"
34
 
 
35
 
#include <boost/dynamic_bitset.hpp>
36
 
#include <list>
37
 
 
38
 
using namespace std;
39
 
 
40
 
namespace drizzled
 
21
#include <drizzled/server_includes.h>
 
22
#include <drizzled/sql_select.h>
 
23
#include <drizzled/error.h>
 
24
#include <drizzled/probes.h>
 
25
#include <drizzled/sql_base.h>
 
26
#include <drizzled/field/timestamp.h>
 
27
 
 
28
/*
 
29
  check that all fields are real fields
 
30
 
 
31
  SYNOPSIS
 
32
    check_fields()
 
33
    session             thread handler
 
34
    items           Items for check
 
35
 
 
36
  RETURN
 
37
    true  Items can't be used in UPDATE
 
38
    false Items are OK
 
39
*/
 
40
 
 
41
static bool check_fields(Session *session, List<Item> &items)
41
42
{
 
43
  List_iterator<Item> it(items);
 
44
  Item *item;
 
45
  Item_field *field;
 
46
 
 
47
  while ((item= it++))
 
48
  {
 
49
    if (!(field= item->filed_for_view_update()))
 
50
    {
 
51
      /* item has name, because it comes from VIEW SELECT list */
 
52
      my_error(ER_NONUPDATEABLE_COLUMN, MYF(0), item->name);
 
53
      return true;
 
54
    }
 
55
    /*
 
56
      we make temporary copy of Item_field, to avoid influence of changing
 
57
      result_field on Item_ref which refer on this field
 
58
    */
 
59
    session->change_item_tree(it.ref(), new Item_field(session, field));
 
60
  }
 
61
  return false;
 
62
}
 
63
 
42
64
 
43
65
/**
44
66
  Re-read record if more columns are needed for error message.
45
67
 
46
68
  If we got a duplicate key error, we want to write an error
47
69
  message containing the value of the duplicate key. If we do not have
48
 
  all fields of the key value in getInsertRecord(), we need to re-read the
 
70
  all fields of the key value in record[0], we need to re-read the
49
71
  record with a proper read_set.
50
72
 
51
73
  @param[in] error   error number
54
76
 
55
77
static void prepare_record_for_error_message(int error, Table *table)
56
78
{
57
 
  Field **field_p= NULL;
58
 
  Field *field= NULL;
59
 
  uint32_t keynr= 0;
 
79
  Field **field_p;
 
80
  Field *field;
 
81
  uint32_t keynr;
 
82
  MY_BITMAP unique_map; /* Fields in offended unique. */
 
83
  my_bitmap_map unique_map_buf[bitmap_buffer_size(MAX_FIELDS)];
60
84
 
61
85
  /*
62
86
    Only duplicate key errors print the key value.
63
87
    If storage engine does always read all columns, we have the value alraedy.
64
88
  */
65
89
  if ((error != HA_ERR_FOUND_DUPP_KEY) ||
66
 
      ! (table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ)))
 
90
      !(table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ))
67
91
    return;
68
92
 
69
93
  /*
70
94
    Get the number of the offended index.
71
95
    We will see MAX_KEY if the engine cannot determine the affected index.
72
96
  */
73
 
  if ((keynr= table->get_dup_key(error)) >= MAX_KEY)
 
97
  if ((keynr= table->file->get_dup_key(error)) >= MAX_KEY)
74
98
    return;
75
99
 
76
100
  /* Create unique_map with all fields used by that index. */
77
 
  boost::dynamic_bitset<> unique_map(table->getShare()->sizeFields()); /* Fields in offended unique. */
78
 
  table->mark_columns_used_by_index_no_reset(keynr, unique_map);
 
101
  bitmap_init(&unique_map, unique_map_buf, table->s->fields, false);
 
102
  table->mark_columns_used_by_index_no_reset(keynr, &unique_map);
79
103
 
80
104
  /* Subtract read_set and write_set. */
81
 
  unique_map-= *table->read_set;
82
 
  unique_map-= *table->write_set;
 
105
  bitmap_subtract(&unique_map, table->read_set);
 
106
  bitmap_subtract(&unique_map, table->write_set);
83
107
 
84
108
  /*
85
109
    If the unique index uses columns that are neither in read_set
86
110
    nor in write_set, we must re-read the record.
87
111
    Otherwise no need to do anything.
88
112
  */
89
 
  if (unique_map.none())
 
113
  if (bitmap_is_clear_all(&unique_map))
90
114
    return;
91
115
 
92
 
  /* Get identifier of last read record into table->cursor->ref. */
93
 
  table->cursor->position(table->getInsertRecord());
 
116
  /* Get identifier of last read record into table->file->ref. */
 
117
  table->file->position(table->record[0]);
94
118
  /* Add all fields used by unique index to read_set. */
95
 
  *table->read_set|= unique_map;
96
 
  /* Read record that is identified by table->cursor->ref. */
97
 
  (void) table->cursor->rnd_pos(table->getUpdateRecord(), table->cursor->ref);
 
119
  bitmap_union(table->read_set, &unique_map);
 
120
  /* Tell the engine about the new set. */
 
121
  table->file->column_bitmaps_signal();
 
122
  /* Read record that is identified by table->file->ref. */
 
123
  (void) table->file->rnd_pos(table->record[1], table->file->ref);
98
124
  /* Copy the newly read columns into the new record. */
99
 
  for (field_p= table->getFields(); (field= *field_p); field_p++)
100
 
  {
101
 
    if (unique_map.test(field->position()))
102
 
    {
103
 
      field->copy_from_tmp(table->getShare()->rec_buff_length);
104
 
    }
105
 
  }
 
125
  for (field_p= table->field; (field= *field_p); field_p++)
 
126
    if (bitmap_is_set(&unique_map, field->field_index))
 
127
      field->copy_from_tmp(table->s->rec_buff_length);
106
128
 
107
129
  return;
108
130
}
117
139
    fields              fields for update
118
140
    values              values of fields for update
119
141
    conds               WHERE clause expression
120
 
    order_num           number of elemen in ORDER BY clause
 
142
    order_num           number of elemen in order_st BY clause
121
143
    order               order_st BY clause list
122
144
    limit               limit clause
123
145
    handle_duplicates   how to handle duplicates
129
151
 
130
152
int mysql_update(Session *session, TableList *table_list,
131
153
                 List<Item> &fields, List<Item> &values, COND *conds,
132
 
                 uint32_t order_num, Order *order,
 
154
                 uint32_t order_num, order_st *order,
133
155
                 ha_rows limit, enum enum_duplicates,
134
156
                 bool ignore)
135
157
{
136
158
  bool          using_limit= limit != HA_POS_ERROR;
137
 
  bool          used_key_is_modified;
138
 
  bool          transactional_table;
139
 
  int           error;
 
159
  bool          safe_update= test(session->options & OPTION_SAFE_UPDATES);
 
160
  bool          used_key_is_modified, transactional_table, will_batch;
 
161
  bool          can_compare_record;
 
162
  int           error, loc_error;
140
163
  uint          used_index= MAX_KEY, dup_key_found;
141
164
  bool          need_sort= true;
 
165
  uint32_t          table_count= 0;
142
166
  ha_rows       updated, found;
143
167
  key_map       old_covering_keys;
144
168
  Table         *table;
145
 
  optimizer::SqlSelect *select= NULL;
146
 
  ReadRecord    info;
 
169
  SQL_SELECT    *select;
 
170
  READ_RECORD   info;
147
171
  Select_Lex    *select_lex= &session->lex->select_lex;
 
172
  bool          need_reopen;
148
173
  uint64_t     id;
149
174
  List<Item> all_fields;
150
 
  Session::killed_state_t killed_status= Session::NOT_KILLED;
 
175
  Session::killed_state killed_status= Session::NOT_KILLED;
151
176
 
152
 
  DRIZZLE_UPDATE_START(session->getQueryString()->c_str());
153
 
  if (session->openTablesLock(table_list))
 
177
  for ( ; ; )
154
178
  {
155
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
156
 
    return 1;
 
179
    if (open_tables(session, &table_list, &table_count, 0))
 
180
      return(1);
 
181
 
 
182
    if (!lock_tables(session, table_list, table_count, &need_reopen))
 
183
      break;
 
184
    if (!need_reopen)
 
185
      return(1);
 
186
    close_tables_for_reopen(session, &table_list);
157
187
  }
158
188
 
 
189
  if (mysql_handle_derived(session->lex, &mysql_derived_prepare) ||
 
190
      (session->fill_derived_tables() &&
 
191
       mysql_handle_derived(session->lex, &mysql_derived_filling)))
 
192
    return(1);
 
193
 
 
194
  DRIZZLE_UPDATE_START();
159
195
  session->set_proc_info("init");
160
196
  table= table_list->table;
161
197
 
162
198
  /* Calculate "table->covering_keys" based on the WHERE */
163
 
  table->covering_keys= table->getShare()->keys_in_use;
164
 
  table->quick_keys.reset();
 
199
  table->covering_keys= table->s->keys_in_use;
 
200
  table->quick_keys.clear_all();
165
201
 
166
202
  if (mysql_prepare_update(session, table_list, &conds, order_num, order))
167
 
  {
168
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
169
 
    return 1;
170
 
  }
 
203
    goto abort;
171
204
 
172
205
  old_covering_keys= table->covering_keys;              // Keys used in WHERE
173
206
  /* Check the fields we are going to modify */
174
207
  if (setup_fields_with_no_wrap(session, 0, fields, MARK_COLUMNS_WRITE, 0, 0))
175
 
  {
176
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
177
 
    return 1;
178
 
  }
179
 
 
 
208
    goto abort;                               /* purecov: inspected */
180
209
  if (table->timestamp_field)
181
210
  {
182
211
    // Don't set timestamp column if this is modified
183
 
    if (table->timestamp_field->isWriteSet())
184
 
    {
 
212
    if (bitmap_is_set(table->write_set,
 
213
                      table->timestamp_field->field_index))
185
214
      table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
186
 
    }
187
215
    else
188
216
    {
189
217
      if (table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_UPDATE ||
190
218
          table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH)
191
 
      {
192
 
        table->setWriteSet(table->timestamp_field->position());
193
 
      }
 
219
        bitmap_set_bit(table->write_set,
 
220
                       table->timestamp_field->field_index);
194
221
    }
195
222
  }
196
223
 
197
224
  if (setup_fields(session, 0, values, MARK_COLUMNS_READ, 0, 0))
198
225
  {
199
226
    free_underlaid_joins(session, select_lex);
200
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
201
 
 
202
 
    return 1;
 
227
    goto abort;                               /* purecov: inspected */
203
228
  }
204
229
 
205
230
  if (select_lex->inner_refs_list.elements &&
206
231
    fix_inner_refs(session, all_fields, select_lex, select_lex->ref_pointer_array))
207
232
  {
208
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
209
 
    return -1;
 
233
    DRIZZLE_UPDATE_END();
 
234
    return(-1);
210
235
  }
211
236
 
212
237
  if (conds)
222
247
    update force the table handler to retrieve write-only fields to be able
223
248
    to compare records and detect data change.
224
249
  */
225
 
  if (table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ) &&
 
250
  if (table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ &&
226
251
      table->timestamp_field &&
227
252
      (table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_UPDATE ||
228
253
       table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH))
229
 
  {
230
 
    *table->read_set|= *table->write_set;
231
 
  }
 
254
    bitmap_union(table->read_set, table->write_set);
232
255
  // Don't count on usage of 'only index' when calculating which key to use
233
 
  table->covering_keys.reset();
234
 
 
235
 
  /* Update the table->cursor->stats.records number */
236
 
  table->cursor->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
237
 
 
238
 
  select= optimizer::make_select(table, 0, 0, conds, 0, &error);
 
256
  table->covering_keys.clear_all();
 
257
 
 
258
  /* Update the table->file->stats.records number */
 
259
  table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
 
260
 
 
261
  select= make_select(table, 0, 0, conds, 0, &error);
239
262
  if (error || !limit ||
240
 
      (select && select->check_quick(session, false, limit)))
 
263
      (select && select->check_quick(session, safe_update, limit)))
241
264
  {
242
265
    delete select;
243
 
    /**
244
 
     * Resetting the Diagnostic area to prevent
245
 
     * lp bug# 439719
246
 
     */
247
 
    session->main_da.reset_diagnostics_area();
248
266
    free_underlaid_joins(session, select_lex);
249
267
    if (error)
250
 
    {
251
 
      DRIZZLE_UPDATE_DONE(1, 0, 0);
252
 
      return 1;
253
 
    }
254
 
    DRIZZLE_UPDATE_DONE(0, 0, 0);
 
268
      goto abort;                               // Error in where
 
269
    DRIZZLE_UPDATE_END();
255
270
    session->my_ok();                           // No matching records
256
 
    return 0;
 
271
    return(0);
257
272
  }
258
273
  if (!select && limit != HA_POS_ERROR)
259
274
  {
260
 
    if ((used_index= optimizer::get_index_for_order(table, order, limit)) != MAX_KEY)
 
275
    if ((used_index= get_index_for_order(table, order, limit)) != MAX_KEY)
261
276
      need_sort= false;
262
277
  }
263
278
  /* If running in safe sql mode, don't allow updates without keys */
264
 
  if (table->quick_keys.none())
 
279
  if (table->quick_keys.is_clear_all())
265
280
  {
266
281
    session->server_status|=SERVER_QUERY_NO_INDEX_USED;
 
282
    if (safe_update && !using_limit)
 
283
    {
 
284
      my_message(ER_UPDATE_WITHOUT_KEY_IN_SAFE_MODE,
 
285
                 ER(ER_UPDATE_WITHOUT_KEY_IN_SAFE_MODE), MYF(0));
 
286
      goto err;
 
287
    }
267
288
  }
268
289
 
269
290
  table->mark_columns_needed_for_update();
274
295
  {
275
296
    used_index= select->quick->index;
276
297
    used_key_is_modified= (!select->quick->unique_key_range() &&
277
 
                          select->quick->is_keys_used(*table->write_set));
 
298
                          select->quick->is_keys_used(table->write_set));
278
299
  }
279
300
  else
280
301
  {
281
302
    used_key_is_modified= 0;
282
303
    if (used_index == MAX_KEY)                  // no index for sort order
283
 
      used_index= table->cursor->key_used_on_scan;
 
304
      used_index= table->file->key_used_on_scan;
284
305
    if (used_index != MAX_KEY)
285
 
      used_key_is_modified= is_key_used(table, used_index, *table->write_set);
 
306
      used_key_is_modified= is_key_used(table, used_index, table->write_set);
286
307
  }
287
308
 
288
309
 
292
313
      We can't update table directly;  We must first search after all
293
314
      matching rows before updating the table!
294
315
    */
295
 
    if (used_index < MAX_KEY && old_covering_keys.test(used_index))
 
316
    if (used_index < MAX_KEY && old_covering_keys.is_set(used_index))
296
317
    {
297
318
      table->key_read=1;
298
319
      table->mark_columns_used_by_index(used_index);
311
332
        NOTE: filesort will call table->prepare_for_position()
312
333
      */
313
334
      uint32_t         length= 0;
314
 
      SortField  *sortorder;
 
335
      SORT_FIELD  *sortorder;
315
336
      ha_rows examined_rows;
316
 
      FileSort filesort(*session);
317
337
 
318
 
      table->sort.io_cache= new internal::IO_CACHE;
 
338
      table->sort.io_cache = new IO_CACHE;
 
339
      memset(table->sort.io_cache, 0, sizeof(IO_CACHE));
319
340
 
320
341
      if (!(sortorder=make_unireg_sortorder(order, &length, NULL)) ||
321
 
          (table->sort.found_records= filesort.run(table, sortorder, length,
322
 
                                                   select, limit, 1,
323
 
                                                   examined_rows)) == HA_POS_ERROR)
 
342
          (table->sort.found_records= filesort(session, table, sortorder, length,
 
343
                                               select, limit, 1,
 
344
                                               &examined_rows))
 
345
          == HA_POS_ERROR)
324
346
      {
325
347
        goto err;
326
348
      }
339
361
        update these in a separate loop based on the pointer.
340
362
      */
341
363
 
342
 
      internal::IO_CACHE tempfile;
343
 
      if (tempfile.open_cached_file(drizzle_tmpdir.c_str(),TEMP_PREFIX, DISK_BUFFER_SIZE, MYF(MY_WME)))
344
 
      {
 
364
      IO_CACHE tempfile;
 
365
      if (open_cached_file(&tempfile, drizzle_tmpdir,TEMP_PREFIX,
 
366
                           DISK_BUFFER_SIZE, MYF(MY_WME)))
345
367
        goto err;
346
 
      }
347
368
 
348
369
      /* If quick select is used, initialize it before retrieving rows. */
349
370
      if (select && select->quick && select->quick->reset())
350
371
        goto err;
351
 
      table->cursor->try_semi_consistent_read(1);
 
372
      table->file->try_semi_consistent_read(1);
352
373
 
353
374
      /*
354
375
        When we get here, we have one of the following options:
362
383
      */
363
384
 
364
385
      if (used_index == MAX_KEY || (select && select->quick))
365
 
      {
366
 
        info.init_read_record(session, table, select, 0, true);
367
 
      }
 
386
        init_read_record(&info,session,table,select,0,1);
368
387
      else
369
 
      {
370
 
        info.init_read_record_idx(session, table, 1, used_index);
371
 
      }
 
388
        init_read_record_idx(&info, session, table, 1, used_index);
372
389
 
373
390
      session->set_proc_info("Searching rows for update");
374
391
      ha_rows tmp_limit= limit;
375
392
 
376
 
      while (not(error= info.read_record(&info)) && not session->getKilled())
 
393
      while (!(error=info.read_record(&info)) && !session->killed)
377
394
      {
378
395
        if (!(select && select->skip_record()))
379
396
        {
380
 
          if (table->cursor->was_semi_consistent_read())
 
397
          if (table->file->was_semi_consistent_read())
381
398
            continue;  /* repeat the read of the same row if it still exists */
382
399
 
383
 
          table->cursor->position(table->getInsertRecord());
384
 
          if (my_b_write(&tempfile,table->cursor->ref,
385
 
                         table->cursor->ref_length))
 
400
          table->file->position(table->record[0]);
 
401
          if (my_b_write(&tempfile,table->file->ref,
 
402
                         table->file->ref_length))
386
403
          {
387
 
            error=1;
388
 
            break;
 
404
            error=1; /* purecov: inspected */
 
405
            break; /* purecov: inspected */
389
406
          }
390
407
          if (!--limit && using_limit)
391
408
          {
394
411
          }
395
412
        }
396
413
        else
397
 
          table->cursor->unlock_row();
 
414
          table->file->unlock_row();
398
415
      }
399
 
      if (session->getKilled() && not error)
 
416
      if (session->killed && !error)
400
417
        error= 1;                               // Aborted
401
418
      limit= tmp_limit;
402
 
      table->cursor->try_semi_consistent_read(0);
403
 
      info.end_read_record();
 
419
      table->file->try_semi_consistent_read(0);
 
420
      end_read_record(&info);
404
421
 
405
422
      /* Change select to use tempfile */
406
423
      if (select)
413
430
      }
414
431
      else
415
432
      {
416
 
        select= new optimizer::SqlSelect;
 
433
        select= new SQL_SELECT;
417
434
        select->head=table;
418
435
      }
419
 
      if (tempfile.reinit_io_cache(internal::READ_CACHE,0L,0,0))
420
 
        error=1;
421
 
      // Read row ptrs from this cursor
422
 
      memcpy(select->file, &tempfile, sizeof(tempfile));
 
436
      if (reinit_io_cache(&tempfile,READ_CACHE,0L,0,0))
 
437
        error=1; /* purecov: inspected */
 
438
      select->file=tempfile;                    // Read row ptrs from this file
423
439
      if (error >= 0)
424
440
        goto err;
425
441
    }
428
444
  }
429
445
 
430
446
  if (ignore)
431
 
    table->cursor->extra(HA_EXTRA_IGNORE_DUP_KEY);
 
447
    table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
432
448
 
433
449
  if (select && select->quick && select->quick->reset())
434
450
    goto err;
435
 
  table->cursor->try_semi_consistent_read(1);
436
 
  info.init_read_record(session, table, select, 0, true);
 
451
  table->file->try_semi_consistent_read(1);
 
452
  init_read_record(&info,session,table,select,0,1);
437
453
 
438
454
  updated= found= 0;
439
455
  /*
448
464
  session->cuted_fields= 0L;
449
465
  session->set_proc_info("Updating");
450
466
 
451
 
  transactional_table= table->cursor->has_transactions();
 
467
  transactional_table= table->file->has_transactions();
452
468
  session->abort_on_warning= test(!ignore);
 
469
  will_batch= !table->file->start_bulk_update();
453
470
 
454
471
  /*
455
472
    Assure that we can use position()
456
473
    if we need to create an error message.
457
474
  */
458
 
  if (table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ))
 
475
  if (table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ)
459
476
    table->prepare_for_position();
460
477
 
461
 
  while (not (error=info.read_record(&info)) && not session->getKilled())
 
478
  /*
 
479
    We can use compare_record() to optimize away updates if
 
480
    the table handler is returning all columns OR if
 
481
    if all updated columns are read
 
482
  */
 
483
  can_compare_record= (!(table->file->ha_table_flags() &
 
484
                         HA_PARTIAL_COLUMN_READ) ||
 
485
                       bitmap_is_subset(table->write_set, table->read_set));
 
486
 
 
487
  while (!(error=info.read_record(&info)) && !session->killed)
462
488
  {
463
 
    if (not (select && select->skip_record()))
 
489
    if (!(select && select->skip_record()))
464
490
    {
465
 
      if (table->cursor->was_semi_consistent_read())
 
491
      if (table->file->was_semi_consistent_read())
466
492
        continue;  /* repeat the read of the same row if it still exists */
467
493
 
468
 
      table->storeRecord();
469
 
      if (fill_record(session, fields, values))
470
 
        break;
 
494
      store_record(table,record[1]);
 
495
      if (fill_record(session, fields, values, 0))
 
496
        break; /* purecov: inspected */
471
497
 
472
498
      found++;
473
499
 
474
 
      if (! table->records_are_comparable() || table->compare_records())
 
500
      if (!can_compare_record || table->compare_record())
475
501
      {
476
 
        /* Non-batched update */
477
 
        error= table->cursor->updateRecord(table->getUpdateRecord(),
478
 
                                            table->getInsertRecord());
479
 
 
480
 
        table->auto_increment_field_not_null= false;
481
 
 
 
502
        if (will_batch)
 
503
        {
 
504
          /*
 
505
            Typically a batched handler can execute the batched jobs when:
 
506
            1) When specifically told to do so
 
507
            2) When it is not a good idea to batch anymore
 
508
            3) When it is necessary to send batch for other reasons
 
509
               (One such reason is when READ's must be performed)
 
510
 
 
511
            1) is covered by exec_bulk_update calls.
 
512
            2) and 3) is handled by the bulk_update_row method.
 
513
 
 
514
            bulk_update_row can execute the updates including the one
 
515
            defined in the bulk_update_row or not including the row
 
516
            in the call. This is up to the handler implementation and can
 
517
            vary from call to call.
 
518
 
 
519
            The dup_key_found reports the number of duplicate keys found
 
520
            in those updates actually executed. It only reports those if
 
521
            the extra call with HA_EXTRA_IGNORE_DUP_KEY have been issued.
 
522
            If this hasn't been issued it returns an error code and can
 
523
            ignore this number. Thus any handler that implements batching
 
524
            for UPDATE IGNORE must also handle this extra call properly.
 
525
 
 
526
            If a duplicate key is found on the record included in this
 
527
            call then it should be included in the count of dup_key_found
 
528
            and error should be set to 0 (only if these errors are ignored).
 
529
          */
 
530
          error= table->file->ha_bulk_update_row(table->record[1],
 
531
                                                 table->record[0],
 
532
                                                 &dup_key_found);
 
533
          limit+= dup_key_found;
 
534
          updated-= dup_key_found;
 
535
        }
 
536
        else
 
537
        {
 
538
          /* Non-batched update */
 
539
          error= table->file->ha_update_row(table->record[1],
 
540
                                            table->record[0]);
 
541
        }
482
542
        if (!error || error == HA_ERR_RECORD_IS_THE_SAME)
483
 
        {
 
543
        {
484
544
          if (error != HA_ERR_RECORD_IS_THE_SAME)
485
545
            updated++;
486
546
          else
487
547
            error= 0;
488
 
        }
489
 
        else if (! ignore ||
490
 
                 table->cursor->is_fatal_error(error, HA_CHECK_DUP_KEY))
491
 
        {
 
548
        }
 
549
        else if (!ignore ||
 
550
                 table->file->is_fatal_error(error, HA_CHECK_DUP_KEY))
 
551
        {
492
552
          /*
493
553
            If (ignore && error is ignorable) we don't have to
494
554
            do anything; otherwise...
495
555
          */
496
556
          myf flags= 0;
497
557
 
498
 
          if (table->cursor->is_fatal_error(error, HA_CHECK_DUP_KEY))
 
558
          if (table->file->is_fatal_error(error, HA_CHECK_DUP_KEY))
499
559
            flags|= ME_FATALERROR; /* Other handler errors are fatal */
500
560
 
501
561
          prepare_record_for_error_message(error, table);
502
 
          table->print_error(error,MYF(flags));
503
 
          error= 1;
504
 
          break;
505
 
        }
 
562
          table->file->print_error(error,MYF(flags));
 
563
          error= 1;
 
564
          break;
 
565
        }
506
566
      }
507
567
 
508
568
      if (!--limit && using_limit)
509
569
      {
510
 
        error= -1;                              // Simulate end of cursor
511
 
        break;
 
570
        /*
 
571
          We have reached end-of-file in most common situations where no
 
572
          batching has occurred and if batching was supposed to occur but
 
573
          no updates were made and finally when the batch execution was
 
574
          performed without error and without finding any duplicate keys.
 
575
          If the batched updates were performed with errors we need to
 
576
          check and if no error but duplicate key's found we need to
 
577
          continue since those are not counted for in limit.
 
578
        */
 
579
        if (will_batch &&
 
580
            ((error= table->file->exec_bulk_update(&dup_key_found)) ||
 
581
             dup_key_found))
 
582
        {
 
583
          if (error)
 
584
          {
 
585
            /* purecov: begin inspected */
 
586
            /*
 
587
              The handler should not report error of duplicate keys if they
 
588
              are ignored. This is a requirement on batching handlers.
 
589
            */
 
590
            prepare_record_for_error_message(error, table);
 
591
            table->file->print_error(error,MYF(0));
 
592
            error= 1;
 
593
            break;
 
594
            /* purecov: end */
 
595
          }
 
596
          /*
 
597
            Either an error was found and we are ignoring errors or there
 
598
            were duplicate keys found. In both cases we need to correct
 
599
            the counters and continue the loop.
 
600
          */
 
601
          limit= dup_key_found; //limit is 0 when we get here so need to +
 
602
          updated-= dup_key_found;
 
603
        }
 
604
        else
 
605
        {
 
606
          error= -1;                            // Simulate end of file
 
607
          break;
 
608
        }
512
609
      }
513
610
    }
514
611
    else
515
 
      table->cursor->unlock_row();
 
612
      table->file->unlock_row();
516
613
    session->row_count++;
517
614
  }
518
615
  dup_key_found= 0;
524
621
    It's assumed that if an error was set in combination with an effective
525
622
    killed status then the error is due to killing.
526
623
  */
527
 
  killed_status= session->getKilled(); // get the status of the volatile
 
624
  killed_status= session->killed; // get the status of the volatile
528
625
  // simulated killing after the loop must be ineffective for binlogging
529
626
  error= (killed_status == Session::NOT_KILLED)?  error : 1;
530
627
 
531
 
  updated-= dup_key_found;
532
 
  table->cursor->try_semi_consistent_read(0);
 
628
  if (error &&
 
629
      will_batch &&
 
630
      (loc_error= table->file->exec_bulk_update(&dup_key_found)))
 
631
    /*
 
632
      An error has occurred when a batched update was performed and returned
 
633
      an error indication. It cannot be an allowed duplicate key error since
 
634
      we require the batching handler to treat this as a normal behavior.
 
635
 
 
636
      Otherwise we simply remove the number of duplicate keys records found
 
637
      in the batched update.
 
638
    */
 
639
  {
 
640
    /* purecov: begin inspected */
 
641
    prepare_record_for_error_message(loc_error, table);
 
642
    table->file->print_error(loc_error,MYF(ME_FATALERROR));
 
643
    error= 1;
 
644
    /* purecov: end */
 
645
  }
 
646
  else
 
647
    updated-= dup_key_found;
 
648
  if (will_batch)
 
649
    table->file->end_bulk_update();
 
650
  table->file->try_semi_consistent_read(0);
533
651
 
534
652
  if (!transactional_table && updated > 0)
535
 
    session->transaction.stmt.markModifiedNonTransData();
 
653
    session->transaction.stmt.modified_non_trans_table= true;
536
654
 
537
 
  info.end_read_record();
 
655
  end_read_record(&info);
538
656
  delete select;
539
657
  session->set_proc_info("end");
540
 
  table->cursor->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
658
  table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
541
659
 
542
660
  /*
543
661
    error < 0 means really no error at all: we processed all rows until the
544
662
    last one without error. error > 0 means an error (e.g. unique key
545
663
    violation and no IGNORE or REPLACE). error == 0 is also an error (if
546
664
    preparing the record or invoking before triggers fails). See
547
 
    autocommitOrRollback(error>=0) and return(error>=0) below.
 
665
    ha_autocommit_or_rollback(error>=0) and return(error>=0) below.
548
666
    Sometimes we want to binlog even if we updated no rows, in case user used
549
667
    it to be sure master and slave are in same state.
550
668
  */
551
 
  if ((error < 0) || session->transaction.stmt.hasModifiedNonTransData())
 
669
  if ((error < 0) || session->transaction.stmt.modified_non_trans_table)
552
670
  {
553
 
    if (session->transaction.stmt.hasModifiedNonTransData())
554
 
      session->transaction.all.markModifiedNonTransData();
 
671
    if (session->transaction.stmt.modified_non_trans_table)
 
672
      session->transaction.all.modified_non_trans_table= true;
555
673
  }
556
 
  assert(transactional_table || !updated || session->transaction.stmt.hasModifiedNonTransData());
 
674
  assert(transactional_table || !updated || session->transaction.stmt.modified_non_trans_table);
557
675
  free_underlaid_joins(session, select_lex);
558
676
 
559
677
  /* If LAST_INSERT_ID(X) was used, report X */
560
678
  id= session->arg_of_last_insert_id_function ?
561
679
    session->first_successful_insert_id_in_prev_stmt : 0;
562
680
 
 
681
  DRIZZLE_UPDATE_END();
563
682
  if (error < 0)
564
683
  {
565
684
    char buff[STRING_BUFFER_USUAL_SIZE];
566
 
    snprintf(buff, sizeof(buff), ER(ER_UPDATE_INFO), (ulong) found, (ulong) updated,
 
685
    sprintf(buff, ER(ER_UPDATE_INFO), (ulong) found, (ulong) updated,
567
686
            (ulong) session->cuted_fields);
568
 
    session->row_count_func= updated;
569
 
    /**
570
 
     * Resetting the Diagnostic area to prevent
571
 
     * lp bug# 439719
572
 
     */
573
 
    session->main_da.reset_diagnostics_area();
574
 
    session->my_ok((ulong) session->row_count_func, found, id, buff);
575
 
    session->status_var.updated_row_count+= session->row_count_func;
 
687
    session->row_count_func=
 
688
      (session->client_capabilities & CLIENT_FOUND_ROWS) ? found : updated;
 
689
    session->my_ok((ulong) session->row_count_func, id, buff);
576
690
  }
577
 
  session->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;              /* calc cuted fields */
 
691
  session->count_cuted_fields= CHECK_FIELD_IGNORE;              /* calc cuted fields */
578
692
  session->abort_on_warning= 0;
579
 
  DRIZZLE_UPDATE_DONE((error >= 0 || session->is_error()), found, updated);
580
 
  return ((error >= 0 || session->is_error()) ? 1 : 0);
 
693
  return((error >= 0 || session->is_error()) ? 1 : 0);
581
694
 
582
695
err:
583
696
  delete select;
585
698
  if (table->key_read)
586
699
  {
587
700
    table->key_read=0;
588
 
    table->cursor->extra(HA_EXTRA_NO_KEYREAD);
 
701
    table->file->extra(HA_EXTRA_NO_KEYREAD);
589
702
  }
590
703
  session->abort_on_warning= 0;
591
704
 
592
 
  DRIZZLE_UPDATE_DONE(1, 0, 0);
593
 
  return 1;
 
705
abort:
 
706
  DRIZZLE_UPDATE_END();
 
707
  return(1);
594
708
}
595
709
 
596
710
/*
601
715
    session                     - thread handler
602
716
    table_list          - global/local table list
603
717
    conds               - conditions
604
 
    order_num           - number of ORDER BY list entries
605
 
    order               - ORDER BY clause list
 
718
    order_num           - number of order_st BY list entries
 
719
    order               - order_st BY clause list
606
720
 
607
721
  RETURN VALUE
608
722
    false OK
609
723
    true  error
610
724
*/
611
725
bool mysql_prepare_update(Session *session, TableList *table_list,
612
 
                         Item **conds, uint32_t order_num, Order *order)
 
726
                         Item **conds, uint32_t order_num, order_st *order)
613
727
{
614
728
  List<Item> all_fields;
615
729
  Select_Lex *select_lex= &session->lex->select_lex;
621
735
                                    table_list,
622
736
                                    &select_lex->leaf_tables,
623
737
                                    false) ||
624
 
      session->setup_conds(table_list, conds) ||
 
738
      setup_conds(session, table_list, select_lex->leaf_tables, conds) ||
625
739
      select_lex->setup_ref_array(session, order_num) ||
626
740
      setup_order(session, select_lex->ref_pointer_array,
627
741
                  table_list, all_fields, all_fields, order))
630
744
  /* Check that we are not using table that we are updating in a sub select */
631
745
  {
632
746
    TableList *duplicate;
633
 
    if ((duplicate= unique_table(table_list, table_list->next_global)))
634
 
    {
635
 
      my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->getTableName());
636
 
      return true;
637
 
    }
638
 
  }
639
 
 
640
 
  return false;
641
 
}
642
 
 
643
 
} /* namespace drizzled */
 
747
    if ((duplicate= unique_table(session, table_list, table_list->next_global, 0)))
 
748
    {
 
749
      update_non_unique_table_error(table_list, "UPDATE", duplicate);
 
750
      my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
 
751
      return true;
 
752
    }
 
753
  }
 
754
 
 
755
  return false;
 
756
}
 
757
 
 
758
 
 
759
/***************************************************************************
 
760
  Update multiple tables from join
 
761
***************************************************************************/
 
762
 
 
763
/*
 
764
  Get table map for list of Item_field
 
765
*/
 
766
 
 
767
static table_map get_table_map(List<Item> *items)
 
768
{
 
769
  List_iterator_fast<Item> item_it(*items);
 
770
  Item_field *item;
 
771
  table_map map= 0;
 
772
 
 
773
  while ((item= (Item_field *) item_it++))
 
774
    map|= item->used_tables();
 
775
  return map;
 
776
}
 
777
 
 
778
 
 
779
/*
 
780
  make update specific preparation and checks after opening tables
 
781
 
 
782
  SYNOPSIS
 
783
    mysql_multi_update_prepare()
 
784
    session         thread handler
 
785
 
 
786
  RETURN
 
787
    false OK
 
788
    true  Error
 
789
*/
 
790
 
 
791
int mysql_multi_update_prepare(Session *session)
 
792
{
 
793
  LEX *lex= session->lex;
 
794
  TableList *table_list= lex->query_tables;
 
795
  TableList *tl, *leaves;
 
796
  List<Item> *fields= &lex->select_lex.item_list;
 
797
  table_map tables_for_update;
 
798
  bool update_view= 0;
 
799
  /*
 
800
    if this multi-update was converted from usual update, here is table
 
801
    counter else junk will be assigned here, but then replaced with real
 
802
    count in open_tables()
 
803
  */
 
804
  uint32_t  table_count= lex->table_count;
 
805
  const bool using_lock_tables= session->locked_tables != 0;
 
806
  bool original_multiupdate= (session->lex->sql_command == SQLCOM_UPDATE_MULTI);
 
807
  bool need_reopen= false;
 
808
 
 
809
 
 
810
  /* following need for prepared statements, to run next time multi-update */
 
811
  session->lex->sql_command= SQLCOM_UPDATE_MULTI;
 
812
 
 
813
reopen_tables:
 
814
 
 
815
  /* open tables and create derived ones, but do not lock and fill them */
 
816
  if (((original_multiupdate || need_reopen) &&
 
817
       open_tables(session, &table_list, &table_count, 0)) ||
 
818
      mysql_handle_derived(lex, &mysql_derived_prepare))
 
819
    return true;
 
820
  /*
 
821
    setup_tables() need for VIEWs. JOIN::prepare() will call setup_tables()
 
822
    second time, but this call will do nothing (there are check for second
 
823
    call in setup_tables()).
 
824
  */
 
825
 
 
826
  if (setup_tables_and_check_access(session, &lex->select_lex.context,
 
827
                                    &lex->select_lex.top_join_list,
 
828
                                    table_list,
 
829
                                    &lex->select_lex.leaf_tables, false))
 
830
    return true;
 
831
 
 
832
  if (setup_fields_with_no_wrap(session, 0, *fields, MARK_COLUMNS_WRITE, 0, 0))
 
833
    return true;
 
834
 
 
835
  if (update_view && check_fields(session, *fields))
 
836
  {
 
837
    return true;
 
838
  }
 
839
 
 
840
  tables_for_update= get_table_map(fields);
 
841
 
 
842
  /*
 
843
    Setup timestamp handling and locking mode
 
844
  */
 
845
  leaves= lex->select_lex.leaf_tables;
 
846
  for (tl= leaves; tl; tl= tl->next_leaf)
 
847
  {
 
848
    Table *table= tl->table;
 
849
    /* Only set timestamp column if this is not modified */
 
850
    if (table->timestamp_field &&
 
851
        bitmap_is_set(table->write_set,
 
852
                      table->timestamp_field->field_index))
 
853
      table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
 
854
 
 
855
    /* if table will be updated then check that it is unique */
 
856
    if (table->map & tables_for_update)
 
857
    {
 
858
      table->mark_columns_needed_for_update();
 
859
      /*
 
860
        If table will be updated we should not downgrade lock for it and
 
861
        leave it as is.
 
862
      */
 
863
    }
 
864
    else
 
865
    {
 
866
      /*
 
867
        If we are using the binary log, we need TL_READ_NO_INSERT to get
 
868
        correct order of statements. Otherwise, we use a TL_READ lock to
 
869
        improve performance.
 
870
      */
 
871
      tl->lock_type= TL_READ;
 
872
      tl->updating= 0;
 
873
      /* Update Table::lock_type accordingly. */
 
874
      if (!tl->placeholder() && !using_lock_tables)
 
875
        tl->table->reginfo.lock_type= tl->lock_type;
 
876
    }
 
877
  }
 
878
 
 
879
  /* now lock and fill tables */
 
880
  if (lock_tables(session, table_list, table_count, &need_reopen))
 
881
  {
 
882
    if (!need_reopen)
 
883
      return true;
 
884
 
 
885
    /*
 
886
      We have to reopen tables since some of them were altered or dropped
 
887
      during lock_tables() or something was done with their triggers.
 
888
      Let us do some cleanups to be able do setup_table() and setup_fields()
 
889
      once again.
 
890
    */
 
891
    List_iterator_fast<Item> it(*fields);
 
892
    Item *item;
 
893
    while ((item= it++))
 
894
      item->cleanup();
 
895
 
 
896
    /* We have to cleanup translation tables of views. */
 
897
    for (TableList *tbl= table_list; tbl; tbl= tbl->next_global)
 
898
      tbl->cleanup_items();
 
899
 
 
900
    close_tables_for_reopen(session, &table_list);
 
901
    goto reopen_tables;
 
902
  }
 
903
 
 
904
  /*
 
905
    Check that we are not using table that we are updating, but we should
 
906
    skip all tables of UPDATE SELECT itself
 
907
  */
 
908
  lex->select_lex.exclude_from_table_unique_test= true;
 
909
  /* We only need SELECT privilege for columns in the values list */
 
910
  for (tl= leaves; tl; tl= tl->next_leaf)
 
911
  {
 
912
    if (tl->lock_type != TL_READ &&
 
913
        tl->lock_type != TL_READ_NO_INSERT)
 
914
    {
 
915
      TableList *duplicate;
 
916
      if ((duplicate= unique_table(session, tl, table_list, 0)))
 
917
      {
 
918
        update_non_unique_table_error(table_list, "UPDATE", duplicate);
 
919
        return true;
 
920
      }
 
921
    }
 
922
  }
 
923
  /*
 
924
    Set exclude_from_table_unique_test value back to false. It is needed for
 
925
    further check in multi_update::prepare whether to use record cache.
 
926
  */
 
927
  lex->select_lex.exclude_from_table_unique_test= false;
 
928
 
 
929
  if (session->fill_derived_tables() &&
 
930
      mysql_handle_derived(lex, &mysql_derived_filling))
 
931
    return true;
 
932
 
 
933
  return (false);
 
934
}
 
935
 
 
936
 
 
937
/*
 
938
  Setup multi-update handling and call SELECT to do the join
 
939
*/
 
940
 
 
941
bool mysql_multi_update(Session *session,
 
942
                        TableList *table_list,
 
943
                        List<Item> *fields,
 
944
                        List<Item> *values,
 
945
                        COND *conds,
 
946
                        uint64_t options,
 
947
                        enum enum_duplicates handle_duplicates, bool ignore,
 
948
                        Select_Lex_Unit *unit, Select_Lex *select_lex)
 
949
{
 
950
  multi_update *result;
 
951
  bool res;
 
952
 
 
953
  if (!(result= new multi_update(table_list,
 
954
                                 session->lex->select_lex.leaf_tables,
 
955
                                 fields, values,
 
956
                                 handle_duplicates, ignore)))
 
957
    return true;
 
958
 
 
959
  session->abort_on_warning= true;
 
960
 
 
961
  List<Item> total_list;
 
962
  res= mysql_select(session, &select_lex->ref_pointer_array,
 
963
                      table_list, select_lex->with_wild,
 
964
                      total_list,
 
965
                      conds, 0, (order_st *) NULL, (order_st *)NULL, (Item *) NULL,
 
966
                      (order_st *)NULL,
 
967
                      options | SELECT_NO_JOIN_CACHE | SELECT_NO_UNLOCK |
 
968
                      OPTION_SETUP_TABLES_DONE,
 
969
                      result, unit, select_lex);
 
970
  res|= session->is_error();
 
971
  if (unlikely(res))
 
972
  {
 
973
    /* If we had a another error reported earlier then this will be ignored */
 
974
    result->send_error(ER_UNKNOWN_ERROR, ER(ER_UNKNOWN_ERROR));
 
975
    result->abort();
 
976
  }
 
977
  delete result;
 
978
  session->abort_on_warning= 0;
 
979
  return false;
 
980
}
 
981
 
 
982
 
 
983
multi_update::multi_update(TableList *table_list,
 
984
                           TableList *leaves_list,
 
985
                           List<Item> *field_list, List<Item> *value_list,
 
986
                           enum enum_duplicates handle_duplicates_arg,
 
987
                           bool ignore_arg)
 
988
  :all_tables(table_list), leaves(leaves_list), update_tables(0),
 
989
   tmp_tables(0), updated(0), found(0), fields(field_list),
 
990
   values(value_list), table_count(0), copy_field(0),
 
991
   handle_duplicates(handle_duplicates_arg), do_update(1), trans_safe(1),
 
992
   transactional_tables(0), ignore(ignore_arg), error_handled(0)
 
993
{}
 
994
 
 
995
 
 
996
/*
 
997
  Connect fields with tables and create list of tables that are updated
 
998
*/
 
999
 
 
1000
int multi_update::prepare(List<Item> &,
 
1001
                          Select_Lex_Unit *)
 
1002
{
 
1003
  TableList *table_ref;
 
1004
  SQL_LIST update;
 
1005
  table_map tables_to_update;
 
1006
  Item_field *item;
 
1007
  List_iterator_fast<Item> field_it(*fields);
 
1008
  List_iterator_fast<Item> value_it(*values);
 
1009
  uint32_t i, max_fields;
 
1010
  uint32_t leaf_table_count= 0;
 
1011
 
 
1012
  session->count_cuted_fields= CHECK_FIELD_WARN;
 
1013
  session->cuted_fields=0L;
 
1014
  session->set_proc_info("updating main table");
 
1015
 
 
1016
  tables_to_update= get_table_map(fields);
 
1017
 
 
1018
  if (!tables_to_update)
 
1019
  {
 
1020
    my_message(ER_NO_TABLES_USED, ER(ER_NO_TABLES_USED), MYF(0));
 
1021
    return(1);
 
1022
  }
 
1023
 
 
1024
  /*
 
1025
    We have to check values after setup_tables to get covering_keys right in
 
1026
    reference tables
 
1027
  */
 
1028
 
 
1029
  if (setup_fields(session, 0, *values, MARK_COLUMNS_READ, 0, 0))
 
1030
    return(1);
 
1031
 
 
1032
  /*
 
1033
    Save tables beeing updated in update_tables
 
1034
    update_table->shared is position for table
 
1035
    Don't use key read on tables that are updated
 
1036
  */
 
1037
 
 
1038
  update.empty();
 
1039
  for (table_ref= leaves; table_ref; table_ref= table_ref->next_leaf)
 
1040
  {
 
1041
    /* TODO: add support of view of join support */
 
1042
    Table *table=table_ref->table;
 
1043
    leaf_table_count++;
 
1044
    if (tables_to_update & table->map)
 
1045
    {
 
1046
      TableList *tl= (TableList*) session->memdup((char*) table_ref,
 
1047
                                                sizeof(*tl));
 
1048
      if (!tl)
 
1049
        return(1);
 
1050
      update.link_in_list((unsigned char*) tl, (unsigned char**) &tl->next_local);
 
1051
      tl->shared= table_count++;
 
1052
      table->no_keyread=1;
 
1053
      table->covering_keys.clear_all();
 
1054
      table->pos_in_table_list= tl;
 
1055
    }
 
1056
  }
 
1057
 
 
1058
 
 
1059
  table_count=  update.elements;
 
1060
  update_tables= (TableList*) update.first;
 
1061
 
 
1062
  tmp_tables = (Table**) session->calloc(sizeof(Table *) * table_count);
 
1063
  tmp_table_param = (Tmp_Table_Param*) session->calloc(sizeof(Tmp_Table_Param) *
 
1064
                                                   table_count);
 
1065
  fields_for_table= (List_item **) session->alloc(sizeof(List_item *) *
 
1066
                                              table_count);
 
1067
  values_for_table= (List_item **) session->alloc(sizeof(List_item *) *
 
1068
                                              table_count);
 
1069
  if (session->is_fatal_error)
 
1070
    return(1);
 
1071
  for (i=0 ; i < table_count ; i++)
 
1072
  {
 
1073
    fields_for_table[i]= new List_item;
 
1074
    values_for_table[i]= new List_item;
 
1075
  }
 
1076
  if (session->is_fatal_error)
 
1077
    return(1);
 
1078
 
 
1079
  /* Split fields into fields_for_table[] and values_by_table[] */
 
1080
 
 
1081
  while ((item= (Item_field *) field_it++))
 
1082
  {
 
1083
    Item *value= value_it++;
 
1084
    uint32_t offset= item->field->table->pos_in_table_list->shared;
 
1085
    fields_for_table[offset]->push_back(item);
 
1086
    values_for_table[offset]->push_back(value);
 
1087
  }
 
1088
  if (session->is_fatal_error)
 
1089
    return(1);
 
1090
 
 
1091
  /* Allocate copy fields */
 
1092
  max_fields=0;
 
1093
  for (i=0 ; i < table_count ; i++)
 
1094
    set_if_bigger(max_fields, fields_for_table[i]->elements + leaf_table_count);
 
1095
  copy_field= new Copy_field[max_fields];
 
1096
  return(session->is_fatal_error != 0);
 
1097
}
 
1098
 
 
1099
 
 
1100
/*
 
1101
  Check if table is safe to update on fly
 
1102
 
 
1103
  SYNOPSIS
 
1104
    safe_update_on_fly()
 
1105
    session                 Thread handler
 
1106
    join_tab            How table is used in join
 
1107
    all_tables          List of tables
 
1108
 
 
1109
  NOTES
 
1110
    We can update the first table in join on the fly if we know that
 
1111
    a row in this table will never be read twice. This is true under
 
1112
    the following conditions:
 
1113
 
 
1114
    - We are doing a table scan and the data is in a separate file (MyISAM) or
 
1115
      if we don't update a clustered key.
 
1116
 
 
1117
    - We are doing a range scan and we don't update the scan key or
 
1118
      the primary key for a clustered table handler.
 
1119
 
 
1120
    - Table is not joined to itself.
 
1121
 
 
1122
    This function gets information about fields to be updated from
 
1123
    the Table::write_set bitmap.
 
1124
 
 
1125
  WARNING
 
1126
    This code is a bit dependent of how make_join_readinfo() works.
 
1127
 
 
1128
  RETURN
 
1129
    0           Not safe to update
 
1130
    1           Safe to update
 
1131
*/
 
1132
 
 
1133
static bool safe_update_on_fly(Session *session, JOIN_TAB *join_tab,
 
1134
                               TableList *table_ref, TableList *all_tables)
 
1135
{
 
1136
  Table *table= join_tab->table;
 
1137
  if (unique_table(session, table_ref, all_tables, 0))
 
1138
    return 0;
 
1139
  switch (join_tab->type) {
 
1140
  case JT_SYSTEM:
 
1141
  case JT_CONST:
 
1142
  case JT_EQ_REF:
 
1143
    return true;                                // At most one matching row
 
1144
  case JT_REF:
 
1145
  case JT_REF_OR_NULL:
 
1146
    return !is_key_used(table, join_tab->ref.key, table->write_set);
 
1147
  case JT_ALL:
 
1148
    /* If range search on index */
 
1149
    if (join_tab->quick)
 
1150
      return !join_tab->quick->is_keys_used(table->write_set);
 
1151
    /* If scanning in clustered key */
 
1152
    if ((table->file->ha_table_flags() & HA_PRIMARY_KEY_IN_READ_INDEX) &&
 
1153
        table->s->primary_key < MAX_KEY)
 
1154
      return !is_key_used(table, table->s->primary_key, table->write_set);
 
1155
    return true;
 
1156
  default:
 
1157
    break;                                      // Avoid compler warning
 
1158
  }
 
1159
  return false;
 
1160
 
 
1161
}
 
1162
 
 
1163
 
 
1164
/*
 
1165
  Initialize table for multi table
 
1166
 
 
1167
  IMPLEMENTATION
 
1168
    - Update first table in join on the fly, if possible
 
1169
    - Create temporary tables to store changed values for all other tables
 
1170
      that are updated (and main_table if the above doesn't hold).
 
1171
*/
 
1172
 
 
1173
bool
 
1174
multi_update::initialize_tables(JOIN *join)
 
1175
{
 
1176
  TableList *table_ref;
 
1177
 
 
1178
  if ((session->options & OPTION_SAFE_UPDATES) && error_if_full_join(join))
 
1179
    return(1);
 
1180
  main_table=join->join_tab->table;
 
1181
  table_to_update= 0;
 
1182
 
 
1183
  /* Any update has at least one pair (field, value) */
 
1184
  assert(fields->elements);
 
1185
 
 
1186
  /* Create a temporary table for keys to all tables, except main table */
 
1187
  for (table_ref= update_tables; table_ref; table_ref= table_ref->next_local)
 
1188
  {
 
1189
    Table *table=table_ref->table;
 
1190
    uint32_t cnt= table_ref->shared;
 
1191
    List<Item> temp_fields;
 
1192
    order_st     group;
 
1193
    Tmp_Table_Param *tmp_param;
 
1194
 
 
1195
    table->mark_columns_needed_for_update();
 
1196
    if (ignore)
 
1197
      table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
 
1198
    if (table == main_table)                    // First table in join
 
1199
    {
 
1200
      if (safe_update_on_fly(session, join->join_tab, table_ref, all_tables))
 
1201
      {
 
1202
        table_to_update= main_table;            // Update table on the fly
 
1203
        continue;
 
1204
      }
 
1205
    }
 
1206
    table->prepare_for_position();
 
1207
 
 
1208
    tmp_param= tmp_table_param+cnt;
 
1209
 
 
1210
    /*
 
1211
      Create a temporary table to store all fields that are changed for this
 
1212
      table. The first field in the temporary table is a pointer to the
 
1213
      original row so that we can find and update it. For the updatable
 
1214
      VIEW a few following fields are rowids of tables used in the CHECK
 
1215
      OPTION condition.
 
1216
    */
 
1217
 
 
1218
    List_iterator_fast<Table> tbl_it(unupdated_check_opt_tables);
 
1219
    Table *tbl= table;
 
1220
    do
 
1221
    {
 
1222
      Field_varstring *field= new Field_varstring(tbl->file->ref_length, 0,
 
1223
                                                  tbl->alias, tbl->s, &my_charset_bin);
 
1224
      if (!field)
 
1225
        return(1);
 
1226
      field->init(tbl);
 
1227
      /*
 
1228
        The field will be converted to varstring when creating tmp table if
 
1229
        table to be updated was created by mysql 4.1. Deny this.
 
1230
      */
 
1231
      Item_field *ifield= new Item_field((Field *) field);
 
1232
      if (!ifield)
 
1233
         return(1);
 
1234
      ifield->maybe_null= 0;
 
1235
      if (temp_fields.push_back(ifield))
 
1236
        return(1);
 
1237
    } while ((tbl= tbl_it++));
 
1238
 
 
1239
    temp_fields.concat(fields_for_table[cnt]);
 
1240
 
 
1241
    /* Make an unique key over the first field to avoid duplicated updates */
 
1242
    memset(&group, 0, sizeof(group));
 
1243
    group.asc= 1;
 
1244
    group.item= (Item**) temp_fields.head_ref();
 
1245
 
 
1246
    tmp_param->quick_group=1;
 
1247
    tmp_param->field_count=temp_fields.elements;
 
1248
    tmp_param->group_parts=1;
 
1249
    tmp_param->group_length= table->file->ref_length;
 
1250
    if (!(tmp_tables[cnt]=create_tmp_table(session,
 
1251
                                           tmp_param,
 
1252
                                           temp_fields,
 
1253
                                           (order_st*) &group, 0, 0,
 
1254
                                           TMP_TABLE_ALL_COLUMNS,
 
1255
                                           HA_POS_ERROR,
 
1256
                                           (char *) "")))
 
1257
      return(1);
 
1258
    tmp_tables[cnt]->file->extra(HA_EXTRA_WRITE_CACHE);
 
1259
  }
 
1260
  return(0);
 
1261
}
 
1262
 
 
1263
 
 
1264
multi_update::~multi_update()
 
1265
{
 
1266
  TableList *table;
 
1267
  for (table= update_tables ; table; table= table->next_local)
 
1268
  {
 
1269
    table->table->no_keyread= table->table->no_cache= 0;
 
1270
    if (ignore)
 
1271
      table->table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
1272
  }
 
1273
 
 
1274
  if (tmp_tables)
 
1275
  {
 
1276
    for (uint32_t cnt = 0; cnt < table_count; cnt++)
 
1277
    {
 
1278
      if (tmp_tables[cnt])
 
1279
      {
 
1280
        tmp_tables[cnt]->free_tmp_table(session);
 
1281
        tmp_table_param[cnt].cleanup();
 
1282
      }
 
1283
    }
 
1284
  }
 
1285
  if (copy_field)
 
1286
    delete [] copy_field;
 
1287
  session->count_cuted_fields= CHECK_FIELD_IGNORE;              // Restore this setting
 
1288
  assert(trans_safe || !updated ||
 
1289
              session->transaction.all.modified_non_trans_table);
 
1290
}
 
1291
 
 
1292
 
 
1293
bool multi_update::send_data(List<Item> &)
 
1294
{
 
1295
  TableList *cur_table;
 
1296
 
 
1297
  for (cur_table= update_tables; cur_table; cur_table= cur_table->next_local)
 
1298
  {
 
1299
    Table *table= cur_table->table;
 
1300
    uint32_t offset= cur_table->shared;
 
1301
    /*
 
1302
      Check if we are using outer join and we didn't find the row
 
1303
      or if we have already updated this row in the previous call to this
 
1304
      function.
 
1305
 
 
1306
      The same row may be presented here several times in a join of type
 
1307
      UPDATE t1 FROM t1,t2 SET t1.a=t2.a
 
1308
 
 
1309
      In this case we will do the update for the first found row combination.
 
1310
      The join algorithm guarantees that we will not find the a row in
 
1311
      t1 several times.
 
1312
    */
 
1313
    if (table->status & (STATUS_NULL_ROW | STATUS_UPDATED))
 
1314
      continue;
 
1315
 
 
1316
    /*
 
1317
      We can use compare_record() to optimize away updates if
 
1318
      the table handler is returning all columns OR if
 
1319
      if all updated columns are read
 
1320
    */
 
1321
    if (table == table_to_update)
 
1322
    {
 
1323
      bool can_compare_record;
 
1324
      can_compare_record= (!(table->file->ha_table_flags() &
 
1325
                             HA_PARTIAL_COLUMN_READ) ||
 
1326
                           bitmap_is_subset(table->write_set,
 
1327
                                            table->read_set));
 
1328
      table->status|= STATUS_UPDATED;
 
1329
      store_record(table,record[1]);
 
1330
      if (fill_record(session, *fields_for_table[offset],
 
1331
                      *values_for_table[offset], 0))
 
1332
        return(1);
 
1333
 
 
1334
      found++;
 
1335
      if (!can_compare_record || table->compare_record())
 
1336
      {
 
1337
        int error;
 
1338
        if (!updated++)
 
1339
        {
 
1340
          /*
 
1341
            Inform the main table that we are going to update the table even
 
1342
            while we may be scanning it.  This will flush the read cache
 
1343
            if it's used.
 
1344
          */
 
1345
          main_table->file->extra(HA_EXTRA_PREPARE_FOR_UPDATE);
 
1346
        }
 
1347
        if ((error=table->file->ha_update_row(table->record[1],
 
1348
                                              table->record[0])) &&
 
1349
            error != HA_ERR_RECORD_IS_THE_SAME)
 
1350
        {
 
1351
          updated--;
 
1352
          if (!ignore ||
 
1353
              table->file->is_fatal_error(error, HA_CHECK_DUP_KEY))
 
1354
          {
 
1355
            /*
 
1356
              If (ignore && error == is ignorable) we don't have to
 
1357
              do anything; otherwise...
 
1358
            */
 
1359
            myf flags= 0;
 
1360
 
 
1361
            if (table->file->is_fatal_error(error, HA_CHECK_DUP_KEY))
 
1362
              flags|= ME_FATALERROR; /* Other handler errors are fatal */
 
1363
 
 
1364
            prepare_record_for_error_message(error, table);
 
1365
            table->file->print_error(error,MYF(flags));
 
1366
            return(1);
 
1367
          }
 
1368
        }
 
1369
        else
 
1370
        {
 
1371
          if (error == HA_ERR_RECORD_IS_THE_SAME)
 
1372
          {
 
1373
            error= 0;
 
1374
            updated--;
 
1375
          }
 
1376
          /* non-transactional or transactional table got modified   */
 
1377
          /* either multi_update class' flag is raised in its branch */
 
1378
          if (table->file->has_transactions())
 
1379
            transactional_tables= 1;
 
1380
          else
 
1381
          {
 
1382
            trans_safe= 0;
 
1383
            session->transaction.stmt.modified_non_trans_table= true;
 
1384
          }
 
1385
        }
 
1386
      }
 
1387
    }
 
1388
    else
 
1389
    {
 
1390
      int error;
 
1391
      Table *tmp_table= tmp_tables[offset];
 
1392
      /*
 
1393
       For updatable VIEW store rowid of the updated table and
 
1394
       rowids of tables used in the CHECK OPTION condition.
 
1395
      */
 
1396
      uint32_t field_num= 0;
 
1397
      List_iterator_fast<Table> tbl_it(unupdated_check_opt_tables);
 
1398
      Table *tbl= table;
 
1399
      do
 
1400
      {
 
1401
        tbl->file->position(tbl->record[0]);
 
1402
        Field_varstring *ref_field=
 
1403
          reinterpret_cast<Field_varstring *>(tmp_table->field[field_num]);
 
1404
        ref_field->store((char *)tbl->file->ref, tbl->file->ref_length,
 
1405
                         &my_charset_bin);
 
1406
        field_num++;
 
1407
      } while ((tbl= tbl_it++));
 
1408
 
 
1409
      /* Store regular updated fields in the row. */
 
1410
      fill_record(session,
 
1411
                  tmp_table->field + 1 + unupdated_check_opt_tables.elements,
 
1412
                  *values_for_table[offset], 1);
 
1413
 
 
1414
      /* Write row, ignoring duplicated updates to a row */
 
1415
      error= tmp_table->file->ha_write_row(tmp_table->record[0]);
 
1416
      if (error != HA_ERR_FOUND_DUPP_KEY && error != HA_ERR_FOUND_DUPP_UNIQUE)
 
1417
      {
 
1418
        if (error &&
 
1419
            create_myisam_from_heap(session, tmp_table,
 
1420
                                         tmp_table_param[offset].start_recinfo,
 
1421
                                         &tmp_table_param[offset].recinfo,
 
1422
                                         error, 1))
 
1423
        {
 
1424
          do_update=0;
 
1425
          return(1);                    // Not a table_is_full error
 
1426
        }
 
1427
        found++;
 
1428
      }
 
1429
    }
 
1430
  }
 
1431
  return(0);
 
1432
}
 
1433
 
 
1434
 
 
1435
void multi_update::send_error(uint32_t errcode,const char *err)
 
1436
{
 
1437
  /* First send error what ever it is ... */
 
1438
  my_error(errcode, MYF(0), err);
 
1439
}
 
1440
 
 
1441
 
 
1442
void multi_update::abort()
 
1443
{
 
1444
  /* the error was handled or nothing deleted and no side effects return */
 
1445
  if (error_handled ||
 
1446
      (!session->transaction.stmt.modified_non_trans_table && !updated))
 
1447
    return;
 
1448
  /*
 
1449
    If all tables that has been updated are trans safe then just do rollback.
 
1450
    If not attempt to do remaining updates.
 
1451
  */
 
1452
 
 
1453
  if (! trans_safe)
 
1454
  {
 
1455
    assert(session->transaction.stmt.modified_non_trans_table);
 
1456
    if (do_update && table_count > 1)
 
1457
    {
 
1458
      /* Add warning here */
 
1459
      /*
 
1460
         todo/fixme: do_update() is never called with the arg 1.
 
1461
         should it change the signature to become argless?
 
1462
      */
 
1463
      do_updates();
 
1464
    }
 
1465
  }
 
1466
  if (session->transaction.stmt.modified_non_trans_table)
 
1467
  {
 
1468
    session->transaction.all.modified_non_trans_table= true;
 
1469
  }
 
1470
  assert(trans_safe || !updated || session->transaction.stmt.modified_non_trans_table);
 
1471
}
 
1472
 
 
1473
 
 
1474
int multi_update::do_updates()
 
1475
{
 
1476
  TableList *cur_table;
 
1477
  int local_error= 0;
 
1478
  ha_rows org_updated;
 
1479
  Table *table, *tmp_table;
 
1480
  List_iterator_fast<Table> check_opt_it(unupdated_check_opt_tables);
 
1481
 
 
1482
  do_update= 0;                                 // Don't retry this function
 
1483
  if (!found)
 
1484
    return(0);
 
1485
  for (cur_table= update_tables; cur_table; cur_table= cur_table->next_local)
 
1486
  {
 
1487
    bool can_compare_record;
 
1488
    uint32_t offset= cur_table->shared;
 
1489
 
 
1490
    table = cur_table->table;
 
1491
    if (table == table_to_update)
 
1492
      continue;                                 // Already updated
 
1493
    org_updated= updated;
 
1494
    tmp_table= tmp_tables[cur_table->shared];
 
1495
    tmp_table->file->extra(HA_EXTRA_CACHE);     // Change to read cache
 
1496
    (void) table->file->ha_rnd_init(0);
 
1497
    table->file->extra(HA_EXTRA_NO_CACHE);
 
1498
 
 
1499
    check_opt_it.rewind();
 
1500
    while(Table *tbl= check_opt_it++)
 
1501
    {
 
1502
      if (tbl->file->ha_rnd_init(1))
 
1503
        goto err;
 
1504
      tbl->file->extra(HA_EXTRA_CACHE);
 
1505
    }
 
1506
 
 
1507
    /*
 
1508
      Setup copy functions to copy fields from temporary table
 
1509
    */
 
1510
    List_iterator_fast<Item> field_it(*fields_for_table[offset]);
 
1511
    Field **field= tmp_table->field +
 
1512
                   1 + unupdated_check_opt_tables.elements; // Skip row pointers
 
1513
    Copy_field *copy_field_ptr= copy_field, *copy_field_end;
 
1514
    for ( ; *field ; field++)
 
1515
    {
 
1516
      Item_field *item= (Item_field* ) field_it++;
 
1517
      (copy_field_ptr++)->set(item->field, *field, 0);
 
1518
    }
 
1519
    copy_field_end=copy_field_ptr;
 
1520
 
 
1521
    if ((local_error = tmp_table->file->ha_rnd_init(1)))
 
1522
      goto err;
 
1523
 
 
1524
    can_compare_record= (!(table->file->ha_table_flags() &
 
1525
                           HA_PARTIAL_COLUMN_READ) ||
 
1526
                         bitmap_is_subset(table->write_set,
 
1527
                                          table->read_set));
 
1528
 
 
1529
    for (;;)
 
1530
    {
 
1531
      if (session->killed && trans_safe)
 
1532
        goto err;
 
1533
      if ((local_error=tmp_table->file->rnd_next(tmp_table->record[0])))
 
1534
      {
 
1535
        if (local_error == HA_ERR_END_OF_FILE)
 
1536
          break;
 
1537
        if (local_error == HA_ERR_RECORD_DELETED)
 
1538
          continue;                             // May happen on dup key
 
1539
        goto err;
 
1540
      }
 
1541
 
 
1542
      /* call rnd_pos() using rowids from temporary table */
 
1543
      check_opt_it.rewind();
 
1544
      Table *tbl= table;
 
1545
      uint32_t field_num= 0;
 
1546
      do
 
1547
      {
 
1548
        Field_varstring *ref_field=
 
1549
          reinterpret_cast<Field_varstring *>(tmp_table->field[field_num]);
 
1550
        if((local_error=
 
1551
              tbl->file->rnd_pos(tbl->record[0],
 
1552
                                (unsigned char *) ref_field->ptr
 
1553
                                 + ref_field->length_bytes)))
 
1554
          goto err;
 
1555
        field_num++;
 
1556
      } while((tbl= check_opt_it++));
 
1557
 
 
1558
      table->status|= STATUS_UPDATED;
 
1559
      store_record(table,record[1]);
 
1560
 
 
1561
      /* Copy data from temporary table to current table */
 
1562
      for (copy_field_ptr=copy_field;
 
1563
           copy_field_ptr != copy_field_end;
 
1564
           copy_field_ptr++)
 
1565
        (*copy_field_ptr->do_copy)(copy_field_ptr);
 
1566
 
 
1567
      if (!can_compare_record || table->compare_record())
 
1568
      {
 
1569
        if ((local_error=table->file->ha_update_row(table->record[1],
 
1570
                                                    table->record[0])) &&
 
1571
            local_error != HA_ERR_RECORD_IS_THE_SAME)
 
1572
        {
 
1573
          if (!ignore ||
 
1574
              table->file->is_fatal_error(local_error, HA_CHECK_DUP_KEY))
 
1575
            goto err;
 
1576
        }
 
1577
        if (local_error != HA_ERR_RECORD_IS_THE_SAME)
 
1578
          updated++;
 
1579
        else
 
1580
          local_error= 0;
 
1581
      }
 
1582
    }
 
1583
 
 
1584
    if (updated != org_updated)
 
1585
    {
 
1586
      if (table->file->has_transactions())
 
1587
        transactional_tables= 1;
 
1588
      else
 
1589
      {
 
1590
        trans_safe= 0;                          // Can't do safe rollback
 
1591
        session->transaction.stmt.modified_non_trans_table= true;
 
1592
      }
 
1593
    }
 
1594
    (void) table->file->ha_rnd_end();
 
1595
    (void) tmp_table->file->ha_rnd_end();
 
1596
    check_opt_it.rewind();
 
1597
    while (Table *tbl= check_opt_it++)
 
1598
        tbl->file->ha_rnd_end();
 
1599
 
 
1600
  }
 
1601
  return(0);
 
1602
 
 
1603
err:
 
1604
  {
 
1605
    prepare_record_for_error_message(local_error, table);
 
1606
    table->file->print_error(local_error,MYF(ME_FATALERROR));
 
1607
  }
 
1608
 
 
1609
  (void) table->file->ha_rnd_end();
 
1610
  (void) tmp_table->file->ha_rnd_end();
 
1611
  check_opt_it.rewind();
 
1612
  while (Table *tbl= check_opt_it++)
 
1613
      tbl->file->ha_rnd_end();
 
1614
 
 
1615
  if (updated != org_updated)
 
1616
  {
 
1617
    if (table->file->has_transactions())
 
1618
      transactional_tables= 1;
 
1619
    else
 
1620
    {
 
1621
      trans_safe= 0;
 
1622
      session->transaction.stmt.modified_non_trans_table= true;
 
1623
    }
 
1624
  }
 
1625
  return(1);
 
1626
}
 
1627
 
 
1628
 
 
1629
/* out: 1 if error, 0 if success */
 
1630
 
 
1631
bool multi_update::send_eof()
 
1632
{
 
1633
  char buff[STRING_BUFFER_USUAL_SIZE];
 
1634
  uint64_t id;
 
1635
  Session::killed_state killed_status= Session::NOT_KILLED;
 
1636
 
 
1637
  session->set_proc_info("updating reference tables");
 
1638
 
 
1639
  /*
 
1640
     Does updates for the last n - 1 tables, returns 0 if ok;
 
1641
     error takes into account killed status gained in do_updates()
 
1642
  */
 
1643
  int local_error = (table_count) ? do_updates() : 0;
 
1644
  /*
 
1645
    if local_error is not set ON until after do_updates() then
 
1646
    later carried out killing should not affect binlogging.
 
1647
  */
 
1648
  killed_status= (local_error == 0)? Session::NOT_KILLED : session->killed;
 
1649
  session->set_proc_info("end");
 
1650
 
 
1651
  /*
 
1652
    Write the SQL statement to the binlog if we updated
 
1653
    rows and we succeeded or if we updated some non
 
1654
    transactional tables.
 
1655
 
 
1656
    The query has to binlog because there's a modified non-transactional table
 
1657
    either from the query's list or via a stored routine: bug#13270,23333
 
1658
  */
 
1659
 
 
1660
  assert(trans_safe || !updated ||
 
1661
              session->transaction.stmt.modified_non_trans_table);
 
1662
  if (local_error == 0 || session->transaction.stmt.modified_non_trans_table)
 
1663
  {
 
1664
    if (session->transaction.stmt.modified_non_trans_table)
 
1665
      session->transaction.all.modified_non_trans_table= true;
 
1666
  }
 
1667
  if (local_error != 0)
 
1668
    error_handled= true; // to force early leave from ::send_error()
 
1669
 
 
1670
  if (local_error > 0) // if the above log write did not fail ...
 
1671
  {
 
1672
    /* Safety: If we haven't got an error before (can happen in do_updates) */
 
1673
    my_message(ER_UNKNOWN_ERROR, "An error occured in multi-table update",
 
1674
               MYF(0));
 
1675
    return true;
 
1676
  }
 
1677
 
 
1678
  id= session->arg_of_last_insert_id_function ?
 
1679
    session->first_successful_insert_id_in_prev_stmt : 0;
 
1680
  sprintf(buff, ER(ER_UPDATE_INFO), (ulong) found, (ulong) updated,
 
1681
          (ulong) session->cuted_fields);
 
1682
  session->row_count_func=
 
1683
    (session->client_capabilities & CLIENT_FOUND_ROWS) ? found : updated;
 
1684
  session->my_ok((ulong) session->row_count_func, id, buff);
 
1685
 
 
1686
  return false;
 
1687
}