← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_update.cc

  • Committer: Brian Aker
  • Date: 2010-06-02 23:40:24 UTC
  • mto: (1578.6.9 explain-drizzle)
  • mto: This revision was merged to the branch mainline in revision 1585.
  • Revision ID: brian@gir-2.local-20100602234024-x26qr3pfhvood1d7
keys and fields partial encapsulation.

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
/*
29
29
#include "drizzled/records.h"
30
30
#include "drizzled/internal/my_sys.h"
31
31
#include "drizzled/internal/iocache.h"
32
 
#include "drizzled/transaction_services.h"
33
 
#include "drizzled/filesort.h"
34
32
 
35
 
#include <boost/dynamic_bitset.hpp>
36
33
#include <list>
37
34
 
38
35
using namespace std;
45
42
 
46
43
  If we got a duplicate key error, we want to write an error
47
44
  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
 
45
  all fields of the key value in record[0], we need to re-read the
49
46
  record with a proper read_set.
50
47
 
51
48
  @param[in] error   error number
54
51
 
55
52
static void prepare_record_for_error_message(int error, Table *table)
56
53
{
57
 
  Field **field_p= NULL;
58
 
  Field *field= NULL;
59
 
  uint32_t keynr= 0;
 
54
  Field **field_p;
 
55
  Field *field;
 
56
  uint32_t keynr;
 
57
  MyBitmap unique_map; /* Fields in offended unique. */
 
58
  my_bitmap_map unique_map_buf[bitmap_buffer_size(MAX_FIELDS)];
60
59
 
61
60
  /*
62
61
    Only duplicate key errors print the key value.
63
62
    If storage engine does always read all columns, we have the value alraedy.
64
63
  */
65
64
  if ((error != HA_ERR_FOUND_DUPP_KEY) ||
66
 
      ! (table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ)))
 
65
      !(table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ)))
67
66
    return;
68
67
 
69
68
  /*
74
73
    return;
75
74
 
76
75
  /* 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);
 
76
  unique_map.init(unique_map_buf, table->getMutableShare()->sizeFields());
 
77
  table->mark_columns_used_by_index_no_reset(keynr, &unique_map);
79
78
 
80
79
  /* Subtract read_set and write_set. */
81
 
  unique_map-= *table->read_set;
82
 
  unique_map-= *table->write_set;
 
80
  bitmap_subtract(&unique_map, table->read_set);
 
81
  bitmap_subtract(&unique_map, table->write_set);
83
82
 
84
83
  /*
85
84
    If the unique index uses columns that are neither in read_set
86
85
    nor in write_set, we must re-read the record.
87
86
    Otherwise no need to do anything.
88
87
  */
89
 
  if (unique_map.none())
 
88
  if (unique_map.isClearAll())
90
89
    return;
91
90
 
92
91
  /* Get identifier of last read record into table->cursor->ref. */
93
 
  table->cursor->position(table->getInsertRecord());
 
92
  table->cursor->position(table->record[0]);
94
93
  /* Add all fields used by unique index to read_set. */
95
 
  *table->read_set|= unique_map;
 
94
  bitmap_union(table->read_set, &unique_map);
96
95
  /* Read record that is identified by table->cursor->ref. */
97
 
  (void) table->cursor->rnd_pos(table->getUpdateRecord(), table->cursor->ref);
 
96
  (void) table->cursor->rnd_pos(table->record[1], table->cursor->ref);
98
97
  /* 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
 
    {
 
98
  for (field_p= table->field; (field= *field_p); field_p++)
 
99
    if (unique_map.isBitSet(field->field_index))
103
100
      field->copy_from_tmp(table->getShare()->rec_buff_length);
104
 
    }
105
 
  }
106
101
 
107
102
  return;
108
103
}
129
124
 
130
125
int mysql_update(Session *session, TableList *table_list,
131
126
                 List<Item> &fields, List<Item> &values, COND *conds,
132
 
                 uint32_t order_num, Order *order,
 
127
                 uint32_t order_num, order_st *order,
133
128
                 ha_rows limit, enum enum_duplicates,
134
129
                 bool ignore)
135
130
{
136
131
  bool          using_limit= limit != HA_POS_ERROR;
137
132
  bool          used_key_is_modified;
138
133
  bool          transactional_table;
 
134
  bool          can_compare_record;
139
135
  int           error;
140
136
  uint          used_index= MAX_KEY, dup_key_found;
141
137
  bool          need_sort= true;
147
143
  Select_Lex    *select_lex= &session->lex->select_lex;
148
144
  uint64_t     id;
149
145
  List<Item> all_fields;
150
 
  Session::killed_state_t killed_status= Session::NOT_KILLED;
 
146
  Session::killed_state killed_status= Session::NOT_KILLED;
151
147
 
152
 
  DRIZZLE_UPDATE_START(session->getQueryString()->c_str());
 
148
  DRIZZLE_UPDATE_START(session->query.c_str());
153
149
  if (session->openTablesLock(table_list))
154
150
  {
155
151
    DRIZZLE_UPDATE_DONE(1, 0, 0);
164
160
  table->quick_keys.reset();
165
161
 
166
162
  if (mysql_prepare_update(session, table_list, &conds, order_num, order))
167
 
  {
168
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
169
 
    return 1;
170
 
  }
 
163
    goto abort;
171
164
 
172
165
  old_covering_keys= table->covering_keys;              // Keys used in WHERE
173
166
  /* Check the fields we are going to modify */
174
167
  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
 
 
 
168
    goto abort;
180
169
  if (table->timestamp_field)
181
170
  {
182
171
    // Don't set timestamp column if this is modified
183
172
    if (table->timestamp_field->isWriteSet())
184
 
    {
185
173
      table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
186
 
    }
187
174
    else
188
175
    {
189
176
      if (table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_UPDATE ||
190
177
          table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH)
191
 
      {
192
 
        table->setWriteSet(table->timestamp_field->position());
193
 
      }
 
178
        table->setWriteSet(table->timestamp_field->field_index);
194
179
    }
195
180
  }
196
181
 
197
182
  if (setup_fields(session, 0, values, MARK_COLUMNS_READ, 0, 0))
198
183
  {
199
184
    free_underlaid_joins(session, select_lex);
200
 
    DRIZZLE_UPDATE_DONE(1, 0, 0);
201
 
 
202
 
    return 1;
 
185
    goto abort;
203
186
  }
204
187
 
205
188
  if (select_lex->inner_refs_list.elements &&
226
209
      table->timestamp_field &&
227
210
      (table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_UPDATE ||
228
211
       table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH))
229
 
  {
230
 
    *table->read_set|= *table->write_set;
231
 
  }
 
212
    bitmap_union(table->read_set, table->write_set);
232
213
  // Don't count on usage of 'only index' when calculating which key to use
233
214
  table->covering_keys.reset();
234
215
 
247
228
    session->main_da.reset_diagnostics_area();
248
229
    free_underlaid_joins(session, select_lex);
249
230
    if (error)
250
 
    {
251
 
      DRIZZLE_UPDATE_DONE(1, 0, 0);
252
 
      return 1;
253
 
    }
 
231
      goto abort;                               // Error in where
254
232
    DRIZZLE_UPDATE_DONE(0, 0, 0);
255
233
    session->my_ok();                           // No matching records
256
234
    return 0;
274
252
  {
275
253
    used_index= select->quick->index;
276
254
    used_key_is_modified= (!select->quick->unique_key_range() &&
277
 
                          select->quick->is_keys_used(*table->write_set));
 
255
                          select->quick->is_keys_used(table->write_set));
278
256
  }
279
257
  else
280
258
  {
282
260
    if (used_index == MAX_KEY)                  // no index for sort order
283
261
      used_index= table->cursor->key_used_on_scan;
284
262
    if (used_index != MAX_KEY)
285
 
      used_key_is_modified= is_key_used(table, used_index, *table->write_set);
 
263
      used_key_is_modified= is_key_used(table, used_index, table->write_set);
286
264
  }
287
265
 
288
266
 
311
289
        NOTE: filesort will call table->prepare_for_position()
312
290
      */
313
291
      uint32_t         length= 0;
314
 
      SortField  *sortorder;
 
292
      SORT_FIELD  *sortorder;
315
293
      ha_rows examined_rows;
316
 
      FileSort filesort(*session);
317
294
 
318
 
      table->sort.io_cache= new internal::IO_CACHE;
 
295
      table->sort.io_cache = new internal::IO_CACHE;
 
296
      memset(table->sort.io_cache, 0, sizeof(internal::IO_CACHE));
319
297
 
320
298
      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)
 
299
          (table->sort.found_records= filesort(session, table, sortorder, length,
 
300
                                               select, limit, 1,
 
301
                                               &examined_rows))
 
302
          == HA_POS_ERROR)
324
303
      {
325
304
        goto err;
326
305
      }
340
319
      */
341
320
 
342
321
      internal::IO_CACHE tempfile;
343
 
      if (tempfile.open_cached_file(drizzle_tmpdir.c_str(),TEMP_PREFIX, DISK_BUFFER_SIZE, MYF(MY_WME)))
344
 
      {
 
322
      if (open_cached_file(&tempfile, drizzle_tmpdir.c_str(),TEMP_PREFIX,
 
323
                           DISK_BUFFER_SIZE, MYF(MY_WME)))
345
324
        goto err;
346
 
      }
347
325
 
348
326
      /* If quick select is used, initialize it before retrieving rows. */
349
327
      if (select && select->quick && select->quick->reset())
373
351
      session->set_proc_info("Searching rows for update");
374
352
      ha_rows tmp_limit= limit;
375
353
 
376
 
      while (not(error= info.read_record(&info)) && not session->getKilled())
 
354
      while (!(error=info.read_record(&info)) && !session->killed)
377
355
      {
378
356
        if (!(select && select->skip_record()))
379
357
        {
380
358
          if (table->cursor->was_semi_consistent_read())
381
359
            continue;  /* repeat the read of the same row if it still exists */
382
360
 
383
 
          table->cursor->position(table->getInsertRecord());
 
361
          table->cursor->position(table->record[0]);
384
362
          if (my_b_write(&tempfile,table->cursor->ref,
385
363
                         table->cursor->ref_length))
386
364
          {
396
374
        else
397
375
          table->cursor->unlock_row();
398
376
      }
399
 
      if (session->getKilled() && not error)
 
377
      if (session->killed && !error)
400
378
        error= 1;                               // Aborted
401
379
      limit= tmp_limit;
402
380
      table->cursor->try_semi_consistent_read(0);
416
394
        select= new optimizer::SqlSelect;
417
395
        select->head=table;
418
396
      }
419
 
      if (tempfile.reinit_io_cache(internal::READ_CACHE,0L,0,0))
 
397
      if (reinit_io_cache(&tempfile,internal::READ_CACHE,0L,0,0))
420
398
        error=1;
421
399
      // Read row ptrs from this cursor
422
400
      memcpy(select->file, &tempfile, sizeof(tempfile));
458
436
  if (table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ))
459
437
    table->prepare_for_position();
460
438
 
461
 
  while (not (error=info.read_record(&info)) && not session->getKilled())
 
439
  /*
 
440
    We can use compare_record() to optimize away updates if
 
441
    the table handler is returning all columns OR if
 
442
    if all updated columns are read
 
443
  */
 
444
  can_compare_record= (!(table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ)) ||
 
445
                       bitmap_is_subset(table->write_set, table->read_set));
 
446
 
 
447
  while (!(error=info.read_record(&info)) && !session->killed)
462
448
  {
463
 
    if (not (select && select->skip_record()))
 
449
    if (!(select && select->skip_record()))
464
450
    {
465
451
      if (table->cursor->was_semi_consistent_read())
466
452
        continue;  /* repeat the read of the same row if it still exists */
471
457
 
472
458
      found++;
473
459
 
474
 
      if (! table->records_are_comparable() || table->compare_records())
 
460
      if (!can_compare_record || table->compare_record())
475
461
      {
476
462
        /* Non-batched update */
477
 
        error= table->cursor->updateRecord(table->getUpdateRecord(),
478
 
                                            table->getInsertRecord());
 
463
        error= table->cursor->updateRecord(table->record[1],
 
464
                                            table->record[0]);
479
465
 
480
466
        table->auto_increment_field_not_null= false;
481
467
 
482
468
        if (!error || error == HA_ERR_RECORD_IS_THE_SAME)
483
 
        {
 
469
        {
484
470
          if (error != HA_ERR_RECORD_IS_THE_SAME)
485
471
            updated++;
486
472
          else
487
473
            error= 0;
488
 
        }
489
 
        else if (! ignore ||
 
474
        }
 
475
        else if (! ignore ||
490
476
                 table->cursor->is_fatal_error(error, HA_CHECK_DUP_KEY))
491
 
        {
 
477
        {
492
478
          /*
493
479
            If (ignore && error is ignorable) we don't have to
494
480
            do anything; otherwise...
499
485
            flags|= ME_FATALERROR; /* Other handler errors are fatal */
500
486
 
501
487
          prepare_record_for_error_message(error, table);
502
 
          table->print_error(error,MYF(flags));
503
 
          error= 1;
504
 
          break;
505
 
        }
 
488
          table->print_error(error,MYF(flags));
 
489
          error= 1;
 
490
          break;
 
491
        }
506
492
      }
507
493
 
508
494
      if (!--limit && using_limit)
524
510
    It's assumed that if an error was set in combination with an effective
525
511
    killed status then the error is due to killing.
526
512
  */
527
 
  killed_status= session->getKilled(); // get the status of the volatile
 
513
  killed_status= session->killed; // get the status of the volatile
528
514
  // simulated killing after the loop must be ineffective for binlogging
529
515
  error= (killed_status == Session::NOT_KILLED)?  error : 1;
530
516
 
572
558
     */
573
559
    session->main_da.reset_diagnostics_area();
574
560
    session->my_ok((ulong) session->row_count_func, found, id, buff);
575
 
    session->status_var.updated_row_count+= session->row_count_func;
576
561
  }
577
 
  session->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;              /* calc cuted fields */
 
562
  session->count_cuted_fields= CHECK_FIELD_IGNORE;              /* calc cuted fields */
578
563
  session->abort_on_warning= 0;
579
564
  DRIZZLE_UPDATE_DONE((error >= 0 || session->is_error()), found, updated);
580
565
  return ((error >= 0 || session->is_error()) ? 1 : 0);
589
574
  }
590
575
  session->abort_on_warning= 0;
591
576
 
 
577
abort:
592
578
  DRIZZLE_UPDATE_DONE(1, 0, 0);
593
579
  return 1;
594
580
}
609
595
    true  error
610
596
*/
611
597
bool mysql_prepare_update(Session *session, TableList *table_list,
612
 
                         Item **conds, uint32_t order_num, Order *order)
 
598
                         Item **conds, uint32_t order_num, order_st *order)
613
599
{
614
600
  List<Item> all_fields;
615
601
  Select_Lex *select_lex= &session->lex->select_lex;
632
618
    TableList *duplicate;
633
619
    if ((duplicate= unique_table(table_list, table_list->next_global)))
634
620
    {
635
 
      my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->getTableName());
 
621
      my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
636
622
      return true;
637
623
    }
638
624
  }