← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_update.cc

Merge Nathan

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