← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_insert.cc

  • Committer: Padraig O'Sullivan
  • Date: 2010-02-10 16:26:01 UTC
  • mto: This revision was merged to the branch mainline in revision 1294.
  • Revision ID: osullivan.padraig@gmail.com-20100210162601-itx2ndl397pc1wr6
Corrected an order of initialization in a few optimizer classes

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_insert.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
/* Insert of records */
18
18
 
19
19
#include "config.h"
20
 
#include <cstdio>
21
20
#include <drizzled/sql_select.h>
22
21
#include <drizzled/show.h>
23
22
#include <drizzled/error.h>
30
29
#include "drizzled/sql_table.h"
31
30
#include "drizzled/pthread_globals.h"
32
31
#include "drizzled/transaction_services.h"
33
 
#include "drizzled/plugin/transactional_storage_engine.h"
34
 
 
35
 
#include "drizzled/table/shell.h"
36
32
 
37
33
namespace drizzled
38
34
{
70
66
 
71
67
  if (fields.elements == 0 && values.elements != 0)
72
68
  {
73
 
    if (values.elements != table->getShare()->sizeFields())
 
69
    if (values.elements != table->s->fields)
74
70
    {
75
71
      my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1L);
76
72
      return -1;
123
119
    if (table->timestamp_field) // Don't automaticly set timestamp if used
124
120
    {
125
121
      if (table->timestamp_field->isWriteSet())
126
 
      {
127
122
        clear_timestamp_auto_bits(table->timestamp_field_type,
128
123
                                  TIMESTAMP_AUTO_SET_ON_INSERT);
129
 
      }
130
124
      else
131
125
      {
132
 
        table->setWriteSet(table->timestamp_field->position());
 
126
        table->setWriteSet(table->timestamp_field->field_index);
133
127
      }
134
128
    }
135
129
  }
170
164
      Unmark the timestamp field so that we can check if this is modified
171
165
      by update_fields
172
166
    */
173
 
    timestamp_mark= table->write_set->test(table->timestamp_field->position());
174
 
    table->write_set->reset(table->timestamp_field->position());
 
167
    timestamp_mark= table->write_set->testAndClear(table->timestamp_field->field_index);
175
168
  }
176
169
 
177
170
  /* Check the fields we are going to modify */
182
175
  {
183
176
    /* Don't set timestamp column if this is modified. */
184
177
    if (table->timestamp_field->isWriteSet())
185
 
    {
186
178
      clear_timestamp_auto_bits(table->timestamp_field_type,
187
179
                                TIMESTAMP_AUTO_SET_ON_UPDATE);
188
 
    }
189
 
 
190
180
    if (timestamp_mark)
191
 
    {
192
 
      table->setWriteSet(table->timestamp_field->position());
193
 
    }
 
181
      table->setWriteSet(table->timestamp_field->field_index);
194
182
  }
195
183
  return 0;
196
184
}
241
229
  uint32_t value_count;
242
230
  ulong counter = 1;
243
231
  uint64_t id;
244
 
  CopyInfo info;
 
232
  COPY_INFO info;
245
233
  Table *table= 0;
246
234
  List_iterator_fast<List_item> its(values_list);
247
235
  List_item *values;
276
264
                           false,
277
265
                           (fields.elements || !value_count ||
278
266
                            (0) != 0), !ignore))
279
 
  {
280
 
    if (table != NULL)
281
 
      table->cursor->ha_release_auto_increment();
282
 
    if (!joins_freed)
283
 
      free_underlaid_joins(session, &session->lex->select_lex);
284
 
    session->abort_on_warning= 0;
285
 
    DRIZZLE_INSERT_DONE(1, 0);
286
 
    return true;
287
 
  }
 
267
    goto abort;
288
268
 
289
269
  /* mysql_prepare_insert set table_list->table if it was not set */
290
270
  table= table_list->table;
315
295
    if (values->elements != value_count)
316
296
    {
317
297
      my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), counter);
318
 
 
319
 
      if (table != NULL)
320
 
        table->cursor->ha_release_auto_increment();
321
 
      if (!joins_freed)
322
 
        free_underlaid_joins(session, &session->lex->select_lex);
323
 
      session->abort_on_warning= 0;
324
 
      DRIZZLE_INSERT_DONE(1, 0);
325
 
 
326
 
      return true;
 
298
      goto abort;
327
299
    }
328
300
    if (setup_fields(session, 0, *values, MARK_COLUMNS_READ, 0, 0))
329
 
    {
330
 
      if (table != NULL)
331
 
        table->cursor->ha_release_auto_increment();
332
 
      if (!joins_freed)
333
 
        free_underlaid_joins(session, &session->lex->select_lex);
334
 
      session->abort_on_warning= 0;
335
 
      DRIZZLE_INSERT_DONE(1, 0);
336
 
      return true;
337
 
    }
 
301
      goto abort;
338
302
  }
339
303
  its.rewind ();
340
304
 
344
308
  /*
345
309
    Fill in the given fields and dump it to the table cursor
346
310
  */
 
311
  memset(&info, 0, sizeof(info));
347
312
  info.ignore= ignore;
348
313
  info.handle_duplicates=duplic;
349
314
  info.update_fields= &update_fields;
354
319
    For single line insert, generate an error if try to set a NOT NULL field
355
320
    to NULL.
356
321
  */
357
 
  session->count_cuted_fields= ignore ? CHECK_FIELD_WARN : CHECK_FIELD_ERROR_FOR_NULL;
358
 
 
 
322
  session->count_cuted_fields= ((values_list.elements == 1 &&
 
323
                                 !ignore) ?
 
324
                                CHECK_FIELD_ERROR_FOR_NULL :
 
325
                                CHECK_FIELD_WARN);
359
326
  session->cuted_fields = 0L;
360
327
  table->next_number_field=table->found_next_number_field;
361
328
 
401
368
    {
402
369
      table->restoreRecordAsDefault();  // Get empty record
403
370
 
404
 
      if (fill_record(session, table->getFields(), *values))
 
371
      if (fill_record(session, table->field, *values))
405
372
      {
406
373
        if (values_list.elements != 1 && ! session->is_error())
407
374
        {
414
381
    }
415
382
 
416
383
    // Release latches in case bulk insert takes a long time
417
 
    plugin::TransactionalStorageEngine::releaseTemporaryLatches(session);
 
384
    plugin::StorageEngine::releaseTemporaryLatches(session);
418
385
 
419
386
    error=write_record(session, table ,&info);
420
387
    if (error)
446
413
    transactional_table= table->cursor->has_transactions();
447
414
 
448
415
    changed= (info.copied || info.deleted || info.updated);
449
 
    if ((changed && error <= 0) || session->transaction.stmt.hasModifiedNonTransData())
 
416
    if ((changed && error <= 0) || session->transaction.stmt.modified_non_trans_table)
450
417
    {
451
 
      if (session->transaction.stmt.hasModifiedNonTransData())
452
 
        session->transaction.all.markModifiedNonTransData();
 
418
      if (session->transaction.stmt.modified_non_trans_table)
 
419
        session->transaction.all.modified_non_trans_table= true;
453
420
    }
454
 
    assert(transactional_table || !changed || session->transaction.stmt.hasModifiedNonTransData());
 
421
    assert(transactional_table || !changed || session->transaction.stmt.modified_non_trans_table);
455
422
 
456
423
  }
457
424
  session->set_proc_info("end");
478
445
    table->cursor->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
479
446
 
480
447
  if (error)
481
 
  {
482
 
    if (table != NULL)
483
 
      table->cursor->ha_release_auto_increment();
484
 
    if (!joins_freed)
485
 
      free_underlaid_joins(session, &session->lex->select_lex);
486
 
    session->abort_on_warning= 0;
487
 
    DRIZZLE_INSERT_DONE(1, 0);
488
 
    return true;
489
 
  }
490
 
 
 
448
    goto abort;
491
449
  if (values_list.elements == 1 && (!(session->options & OPTION_WARNINGS) ||
492
450
                                    !session->cuted_fields))
493
451
  {
499
457
  {
500
458
    char buff[160];
501
459
    if (ignore)
502
 
      snprintf(buff, sizeof(buff), ER(ER_INSERT_INFO), (ulong) info.records,
 
460
      sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
503
461
              (ulong) (info.records - info.copied), (ulong) session->cuted_fields);
504
462
    else
505
 
      snprintf(buff, sizeof(buff), ER(ER_INSERT_INFO), (ulong) info.records,
 
463
      sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
506
464
              (ulong) (info.deleted + info.updated), (ulong) session->cuted_fields);
507
465
    session->row_count_func= info.copied + info.deleted + info.updated;
508
466
    session->my_ok((ulong) session->row_count_func,
509
467
                   info.copied + info.deleted + info.touched, id, buff);
510
468
  }
511
 
  session->status_var.inserted_row_count+= session->row_count_func;
512
469
  session->abort_on_warning= 0;
513
470
  DRIZZLE_INSERT_DONE(0, session->row_count_func);
514
 
 
515
471
  return false;
 
472
 
 
473
abort:
 
474
  if (table != NULL)
 
475
    table->cursor->ha_release_auto_increment();
 
476
  if (!joins_freed)
 
477
    free_underlaid_joins(session, &session->lex->select_lex);
 
478
  session->abort_on_warning= 0;
 
479
  DRIZZLE_INSERT_DONE(1, 0);
 
480
  return true;
516
481
}
517
482
 
518
483
 
706
671
 
707
672
static int last_uniq_key(Table *table,uint32_t keynr)
708
673
{
709
 
  while (++keynr < table->getShare()->sizeKeys())
 
674
  while (++keynr < table->s->keys)
710
675
    if (table->key_info[keynr].flags & HA_NOSAME)
711
676
      return 0;
712
677
  return 1;
721
686
     write_record()
722
687
      session   - thread context
723
688
      table - table to which record should be written
724
 
      info  - CopyInfo structure describing handling of duplicates
 
689
      info  - COPY_INFO structure describing handling of duplicates
725
690
              and which is used for counting number of records inserted
726
691
              and deleted.
727
692
 
731
696
    then both on update triggers will work instead. Similarly both on
732
697
    delete triggers will be invoked if we will delete conflicting records.
733
698
 
734
 
    Sets session->transaction.stmt.modified_non_trans_data to true if table which is updated didn't have
 
699
    Sets session->transaction.stmt.modified_non_trans_table to true if table which is updated didn't have
735
700
    transactions.
736
701
 
737
702
  RETURN VALUE
740
705
*/
741
706
 
742
707
 
743
 
int write_record(Session *session, Table *table,CopyInfo *info)
 
708
int write_record(Session *session, Table *table,COPY_INFO *info)
744
709
{
745
710
  int error;
746
 
  std::vector<unsigned char> key;
747
 
  boost::dynamic_bitset<> *save_read_set, *save_write_set;
 
711
  char *key=0;
 
712
  MyBitmap *save_read_set, *save_write_set;
748
713
  uint64_t prev_insert_id= table->cursor->next_insert_id;
749
714
  uint64_t insert_id_for_cur_row= 0;
750
715
 
755
720
 
756
721
  if (info->handle_duplicates == DUP_REPLACE || info->handle_duplicates == DUP_UPDATE)
757
722
  {
758
 
    while ((error=table->cursor->insertRecord(table->getInsertRecord())))
 
723
    while ((error=table->cursor->ha_write_row(table->record[0])))
759
724
    {
760
725
      uint32_t key_nr;
761
726
      /*
798
763
      */
799
764
      if (info->handle_duplicates == DUP_REPLACE &&
800
765
          table->next_number_field &&
801
 
          key_nr == table->getShare()->next_number_index &&
 
766
          key_nr == table->s->next_number_index &&
802
767
          (insert_id_for_cur_row > 0))
803
768
        goto err;
804
769
      if (table->cursor->getEngine()->check_flag(HTON_BIT_DUPLICATE_POS))
805
770
      {
806
 
        if (table->cursor->rnd_pos(table->getUpdateRecord(),table->cursor->dup_ref))
 
771
        if (table->cursor->rnd_pos(table->record[1],table->cursor->dup_ref))
807
772
          goto err;
808
773
      }
809
774
      else
814
779
          goto err;
815
780
        }
816
781
 
817
 
        if (not key.size())
 
782
        if (!key)
818
783
        {
819
 
          key.resize(table->getShare()->max_unique_length);
 
784
          if (!(key=(char*) malloc(table->s->max_unique_length)))
 
785
          {
 
786
            error=ENOMEM;
 
787
            goto err;
 
788
          }
820
789
        }
821
 
        key_copy(&key[0], table->getInsertRecord(), table->key_info+key_nr, 0);
822
 
        if ((error=(table->cursor->index_read_idx_map(table->getUpdateRecord(),key_nr,
823
 
                                                    &key[0], HA_WHOLE_KEY,
 
790
        key_copy((unsigned char*) key,table->record[0],table->key_info+key_nr,0);
 
791
        if ((error=(table->cursor->index_read_idx_map(table->record[1],key_nr,
 
792
                                                    (unsigned char*) key, HA_WHOLE_KEY,
824
793
                                                    HA_READ_KEY_EXACT))))
825
794
          goto err;
826
795
      }
831
800
          that matches, is updated. If update causes a conflict again,
832
801
          an error is returned
833
802
        */
834
 
        assert(table->insert_values.size());
 
803
        assert(table->insert_values != NULL);
835
804
        table->storeRecordAsInsert();
836
805
        table->restoreRecord();
837
806
        assert(info->update_fields->elements ==
846
815
          table->cursor->adjust_next_insert_id_after_explicit_value(
847
816
            table->next_number_field->val_int());
848
817
        info->touched++;
849
 
 
850
 
        if (! table->records_are_comparable() || table->compare_records())
 
818
        if ((table->cursor->getEngine()->check_flag(HTON_BIT_PARTIAL_COLUMN_READ) &&
 
819
             !bitmap_is_subset(table->write_set, table->read_set)) ||
 
820
            table->compare_record())
851
821
        {
852
 
          if ((error=table->cursor->updateRecord(table->getUpdateRecord(),
853
 
                                                table->getInsertRecord())) &&
 
822
          if ((error=table->cursor->ha_update_row(table->record[1],
 
823
                                                table->record[0])) &&
854
824
              error != HA_ERR_RECORD_IS_THE_SAME)
855
825
          {
856
826
            if (info->ignore &&
904
874
            (table->timestamp_field_type == TIMESTAMP_NO_AUTO_SET ||
905
875
             table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH))
906
876
        {
907
 
          if ((error=table->cursor->updateRecord(table->getUpdateRecord(),
908
 
                                                table->getInsertRecord())) &&
 
877
          if ((error=table->cursor->ha_update_row(table->record[1],
 
878
                                                table->record[0])) &&
909
879
              error != HA_ERR_RECORD_IS_THE_SAME)
910
880
            goto err;
911
881
          if (error != HA_ERR_RECORD_IS_THE_SAME)
921
891
        }
922
892
        else
923
893
        {
924
 
          if ((error=table->cursor->deleteRecord(table->getUpdateRecord())))
 
894
          if ((error=table->cursor->ha_delete_row(table->record[1])))
925
895
            goto err;
926
896
          info->deleted++;
927
897
          if (!table->cursor->has_transactions())
928
 
            session->transaction.stmt.markModifiedNonTransData();
 
898
            session->transaction.stmt.modified_non_trans_table= true;
929
899
          /* Let us attempt do write_row() once more */
930
900
        }
931
901
      }
937
907
    */
938
908
    if (table->read_set != save_read_set ||
939
909
        table->write_set != save_write_set)
940
 
      table->column_bitmaps_set(*save_read_set, *save_write_set);
 
910
      table->column_bitmaps_set(save_read_set, save_write_set);
941
911
  }
942
 
  else if ((error=table->cursor->insertRecord(table->getInsertRecord())))
 
912
  else if ((error=table->cursor->ha_write_row(table->record[0])))
943
913
  {
944
914
    if (!info->ignore ||
945
915
        table->cursor->is_fatal_error(error, HA_CHECK_DUP))
953
923
  session->record_first_successful_insert_id_in_cur_stmt(table->cursor->insert_id_for_cur_row);
954
924
 
955
925
gok_or_after_err:
 
926
  if (key)
 
927
    free(key);
956
928
  if (!table->cursor->has_transactions())
957
 
    session->transaction.stmt.markModifiedNonTransData();
 
929
    session->transaction.stmt.modified_non_trans_table= true;
958
930
  return(0);
959
931
 
960
932
err:
966
938
 
967
939
before_err:
968
940
  table->cursor->restore_auto_increment(prev_insert_id);
969
 
  table->column_bitmaps_set(*save_read_set, *save_write_set);
970
 
  return 1;
 
941
  if (key)
 
942
    free(key);
 
943
  table->column_bitmaps_set(save_read_set, save_write_set);
 
944
  return(1);
971
945
}
972
946
 
973
947
 
980
954
{
981
955
  int err= 0;
982
956
 
983
 
  for (Field **field=entry->getFields() ; *field ; field++)
 
957
  for (Field **field=entry->field ; *field ; field++)
984
958
  {
985
959
    if (((*field)->isWriteSet()) == false)
986
960
    {
1068
1042
                             List<Item> *update_fields,
1069
1043
                             List<Item> *update_values,
1070
1044
                             enum_duplicates duplic,
1071
 
                             bool ignore_check_option_errors) :
1072
 
  table_list(table_list_par), table(table_par), fields(fields_par),
1073
 
  autoinc_value_of_last_inserted_row(0),
1074
 
  insert_into_view(table_list_par && 0 != 0)
 
1045
                             bool ignore_check_option_errors)
 
1046
  :table_list(table_list_par), table(table_par), fields(fields_par),
 
1047
   autoinc_value_of_last_inserted_row(0),
 
1048
   insert_into_view(table_list_par && 0 != 0)
1075
1049
{
 
1050
  memset(&info, 0, sizeof(info));
1076
1051
  info.handle_duplicates= duplic;
1077
1052
  info.ignore= ignore_check_option_errors;
1078
1053
  info.update_fields= update_fields;
1278
1253
    return(1);
1279
1254
 
1280
1255
  // Release latches in case bulk insert takes a long time
1281
 
  plugin::TransactionalStorageEngine::releaseTemporaryLatches(session);
 
1256
  plugin::StorageEngine::releaseTemporaryLatches(session);
1282
1257
 
1283
1258
  error= write_record(session, table, &info);
1284
 
  table->auto_increment_field_not_null= false;
1285
1259
 
1286
1260
  if (!error)
1287
1261
  {
1322
1296
  if (fields->elements)
1323
1297
    fill_record(session, *fields, values, true);
1324
1298
  else
1325
 
    fill_record(session, table->getFields(), values, true);
 
1299
    fill_record(session, table->field, values, true);
1326
1300
}
1327
1301
 
1328
1302
void select_insert::send_error(uint32_t errcode,const char *err)
1350
1324
  {
1351
1325
    /*
1352
1326
      We must invalidate the table in the query cache before binlog writing
1353
 
      and autocommitOrRollback.
 
1327
      and ha_autocommit_or_rollback.
1354
1328
    */
1355
 
    if (session->transaction.stmt.hasModifiedNonTransData())
1356
 
      session->transaction.all.markModifiedNonTransData();
 
1329
    if (session->transaction.stmt.modified_non_trans_table)
 
1330
      session->transaction.all.modified_non_trans_table= true;
1357
1331
  }
1358
1332
  assert(trans_table || !changed ||
1359
 
              session->transaction.stmt.hasModifiedNonTransData());
 
1333
              session->transaction.stmt.modified_non_trans_table);
1360
1334
 
1361
1335
  table->cursor->ha_release_auto_increment();
1362
1336
 
1368
1342
  }
1369
1343
  char buff[160];
1370
1344
  if (info.ignore)
1371
 
    snprintf(buff, sizeof(buff), ER(ER_INSERT_INFO), (ulong) info.records,
 
1345
    sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
1372
1346
            (ulong) (info.records - info.copied), (ulong) session->cuted_fields);
1373
1347
  else
1374
 
    snprintf(buff, sizeof(buff), ER(ER_INSERT_INFO), (ulong) info.records,
 
1348
    sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
1375
1349
            (ulong) (info.deleted+info.updated), (ulong) session->cuted_fields);
1376
1350
  session->row_count_func= info.copied + info.deleted + info.updated;
1377
1351
 
1382
1356
     (info.copied ? autoinc_value_of_last_inserted_row : 0));
1383
1357
  session->my_ok((ulong) session->row_count_func,
1384
1358
                 info.copied + info.deleted + info.touched, id, buff);
1385
 
  session->status_var.inserted_row_count+= session->row_count_func; 
1386
1359
  DRIZZLE_INSERT_SELECT_DONE(0, session->row_count_func);
1387
1360
  return 0;
1388
1361
}
1419
1392
    changed= (info.copied || info.deleted || info.updated);
1420
1393
    transactional_table= table->cursor->has_transactions();
1421
1394
    assert(transactional_table || !changed ||
1422
 
                session->transaction.stmt.hasModifiedNonTransData());
 
1395
                session->transaction.stmt.modified_non_trans_table);
1423
1396
    table->cursor->ha_release_auto_increment();
1424
1397
  }
1425
1398
 
1452
1425
      items        in     List of items which should be used to produce rest
1453
1426
                          of fields for the table (corresponding fields will
1454
1427
                          be added to the end of alter_info->create_list)
1455
 
      lock         out    Pointer to the DrizzleLock object for table created
 
1428
      lock         out    Pointer to the DRIZZLE_LOCK object for table created
1456
1429
                          (or open temporary table) will be returned in this
1457
1430
                          parameter. Since this table is not included in
1458
1431
                          Session::lock caller is responsible for explicitly
1480
1453
 
1481
1454
static Table *create_table_from_items(Session *session, HA_CREATE_INFO *create_info,
1482
1455
                                      TableList *create_table,
1483
 
                                      message::Table &table_proto,
 
1456
                                      message::Table *table_proto,
1484
1457
                                      AlterInfo *alter_info,
1485
1458
                                      List<Item> *items,
1486
1459
                                      bool is_if_not_exists,
1487
 
                                      DrizzleLock **lock,
1488
 
                                      TableIdentifier &identifier)
 
1460
                                      DRIZZLE_LOCK **lock)
1489
1461
{
1490
 
  TableShare share(message::Table::INTERNAL);
 
1462
  Table tmp_table;              // Used during 'CreateField()'
 
1463
  TableShare share;
 
1464
  Table *table= 0;
1491
1465
  uint32_t select_field_count= items->elements;
1492
1466
  /* Add selected items to field list */
1493
1467
  List_iterator_fast<Item> it(*items);
1495
1469
  Field *tmp_field;
1496
1470
  bool not_used;
1497
1471
 
1498
 
  if (not (identifier.isTmp()) && create_table->table->db_stat)
 
1472
  bool lex_identified_temp_table= (table_proto->type() == message::Table::TEMPORARY);
 
1473
 
 
1474
  if (!(lex_identified_temp_table) &&
 
1475
      create_table->table->db_stat)
1499
1476
  {
1500
1477
    /* Table already exists and was open at openTablesLock() stage. */
1501
1478
    if (is_if_not_exists)
1503
1480
      create_info->table_existed= 1;            // Mark that table existed
1504
1481
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
1505
1482
                          ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
1506
 
                          create_table->getTableName());
 
1483
                          create_table->table_name);
1507
1484
      return create_table->table;
1508
1485
    }
1509
1486
 
1510
 
    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->getTableName());
 
1487
    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
1511
1488
    return NULL;
1512
1489
  }
1513
1490
 
 
1491
  tmp_table.alias= 0;
 
1492
  tmp_table.timestamp_field= 0;
 
1493
  tmp_table.s= &share;
 
1494
 
 
1495
  tmp_table.s->db_create_options=0;
 
1496
  tmp_table.s->blob_ptr_size= portable_sizeof_char_ptr;
 
1497
  tmp_table.s->db_low_byte_first=
 
1498
        test(create_info->db_type == myisam_engine ||
 
1499
             create_info->db_type == heap_engine);
 
1500
  tmp_table.null_row= false;
 
1501
  tmp_table.maybe_null= false;
 
1502
 
 
1503
  while ((item=it++))
1514
1504
  {
1515
 
    table::Shell tmp_table(share);              // Used during 'CreateField()'
1516
 
    tmp_table.timestamp_field= 0;
1517
 
 
1518
 
    tmp_table.getMutableShare()->db_create_options= 0;
1519
 
    tmp_table.getMutableShare()->blob_ptr_size= portable_sizeof_char_ptr;
1520
 
 
1521
 
    if (not table_proto.engine().name().compare("MyISAM"))
1522
 
      tmp_table.getMutableShare()->db_low_byte_first= true;
1523
 
    else if (not table_proto.engine().name().compare("MEMORY"))
1524
 
      tmp_table.getMutableShare()->db_low_byte_first= true;
1525
 
 
1526
 
    tmp_table.null_row= false;
1527
 
    tmp_table.maybe_null= false;
1528
 
 
1529
 
    tmp_table.in_use= session;
1530
 
 
1531
 
    while ((item=it++))
1532
 
    {
1533
 
      CreateField *cr_field;
1534
 
      Field *field, *def_field;
1535
 
      if (item->type() == Item::FUNC_ITEM)
1536
 
      {
1537
 
        if (item->result_type() != STRING_RESULT)
1538
 
        {
1539
 
          field= item->tmp_table_field(&tmp_table);
1540
 
        }
1541
 
        else
1542
 
        {
1543
 
          field= item->tmp_table_field_from_field_type(&tmp_table, 0);
1544
 
        }
1545
 
      }
 
1505
    CreateField *cr_field;
 
1506
    Field *field, *def_field;
 
1507
    if (item->type() == Item::FUNC_ITEM)
 
1508
      if (item->result_type() != STRING_RESULT)
 
1509
        field= item->tmp_table_field(&tmp_table);
1546
1510
      else
1547
 
      {
1548
 
        field= create_tmp_field(session, &tmp_table, item, item->type(),
1549
 
                                (Item ***) 0, &tmp_field, &def_field, false,
1550
 
                                false, false, 0);
1551
 
      }
1552
 
 
1553
 
      if (!field ||
1554
 
          !(cr_field=new CreateField(field,(item->type() == Item::FIELD_ITEM ?
1555
 
                                            ((Item_field *)item)->field :
1556
 
                                            (Field*) 0))))
1557
 
      {
1558
 
        return NULL;
1559
 
      }
1560
 
 
1561
 
      if (item->maybe_null)
1562
 
      {
1563
 
        cr_field->flags &= ~NOT_NULL_FLAG;
1564
 
      }
1565
 
 
1566
 
      alter_info->create_list.push_back(cr_field);
1567
 
    }
 
1511
        field= item->tmp_table_field_from_field_type(&tmp_table, 0);
 
1512
    else
 
1513
      field= create_tmp_field(session, &tmp_table, item, item->type(),
 
1514
                              (Item ***) 0, &tmp_field, &def_field, 0, 0, 0, 0,
 
1515
                              0);
 
1516
    if (!field ||
 
1517
        !(cr_field=new CreateField(field,(item->type() == Item::FIELD_ITEM ?
 
1518
                                           ((Item_field *)item)->field :
 
1519
                                           (Field*) 0))))
 
1520
      return NULL;
 
1521
    if (item->maybe_null)
 
1522
      cr_field->flags &= ~NOT_NULL_FLAG;
 
1523
    alter_info->create_list.push_back(cr_field);
1568
1524
  }
1569
1525
 
 
1526
  TableIdentifier identifier(create_table->db,
 
1527
                             create_table->table_name,
 
1528
                             lex_identified_temp_table ?  TEMP_TABLE :
 
1529
                             NO_TMP_TABLE);
 
1530
 
 
1531
 
1570
1532
  /*
1571
1533
    Create and lock table.
1572
1534
 
1574
1536
    creating base table on which name we have exclusive lock. So code below
1575
1537
    should not cause deadlocks or races.
1576
1538
  */
1577
 
  Table *table= 0;
1578
1539
  {
1579
 
    if (not mysql_create_table_no_lock(session,
1580
 
                                       identifier,
1581
 
                                       create_info,
1582
 
                                       table_proto,
1583
 
                                       alter_info,
1584
 
                                       false,
1585
 
                                       select_field_count,
1586
 
                                       is_if_not_exists))
 
1540
    if (!mysql_create_table_no_lock(session,
 
1541
                                    identifier,
 
1542
                                    create_info,
 
1543
                                    table_proto,
 
1544
                                    alter_info,
 
1545
                                    false,
 
1546
                                    select_field_count,
 
1547
                                    is_if_not_exists))
1587
1548
    {
1588
 
      if (create_info->table_existed && not identifier.isTmp())
 
1549
      if (create_info->table_existed &&
 
1550
          !(lex_identified_temp_table))
1589
1551
      {
1590
1552
        /*
1591
1553
          This means that someone created table underneath server
1592
1554
          or it was created via different mysqld front-end to the
1593
1555
          cluster. We don't have much options but throw an error.
1594
1556
        */
1595
 
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->getTableName());
 
1557
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
1596
1558
        return NULL;
1597
1559
      }
1598
1560
 
1599
 
      if (not identifier.isTmp())
 
1561
      if (!(lex_identified_temp_table))
1600
1562
      {
1601
 
        /* CREATE TABLE... has found that the table already exists for insert and is adapting to use it */
1602
 
        boost::mutex::scoped_lock scopedLock(table::Cache::singleton().mutex());
1603
 
 
1604
 
        if (create_table->table)
 
1563
        pthread_mutex_lock(&LOCK_open); /* CREATE TABLE... has found that the table already exists for insert and is adapting to use it */
 
1564
        if (session->reopen_name_locked_table(create_table, false))
1605
1565
        {
1606
 
          table::Concurrent *concurrent_table= static_cast<table::Concurrent *>(create_table->table);
1607
 
 
1608
 
          if (concurrent_table->reopen_name_locked_table(create_table, session))
1609
 
          {
1610
 
            plugin::StorageEngine::dropTable(*session, identifier);
1611
 
          }
1612
 
          else
1613
 
          {
1614
 
            table= create_table->table;
1615
 
          }
 
1566
          quick_rm_table(*session, identifier);
1616
1567
        }
1617
1568
        else
1618
 
        {
1619
 
          plugin::StorageEngine::dropTable(*session, identifier);
1620
 
        }
 
1569
          table= create_table->table;
 
1570
        pthread_mutex_unlock(&LOCK_open);
1621
1571
      }
1622
1572
      else
1623
1573
      {
1624
 
        if (not (table= session->openTable(create_table, (bool*) 0,
1625
 
                                           DRIZZLE_OPEN_TEMPORARY_ONLY)) &&
1626
 
            not create_info->table_existed)
 
1574
        if (!(table= session->openTable(create_table, (bool*) 0,
 
1575
                                         DRIZZLE_OPEN_TEMPORARY_ONLY)) &&
 
1576
            !create_info->table_existed)
1627
1577
        {
1628
1578
          /*
1629
1579
            This shouldn't happen as creation of temporary table should make
1630
1580
            it preparable for open. But let us do close_temporary_table() here
1631
1581
            just in case.
1632
1582
          */
1633
 
          session->drop_temporary_table(identifier);
 
1583
          session->drop_temporary_table(create_table);
1634
1584
        }
1635
1585
      }
1636
1586
    }
1637
 
    if (not table)                                   // open failed
 
1587
    if (!table)                                   // open failed
1638
1588
      return NULL;
1639
1589
  }
1640
1590
 
1641
1591
  table->reginfo.lock_type=TL_WRITE;
1642
 
  if (! ((*lock)= session->lockTables(&table, 1, DRIZZLE_LOCK_IGNORE_FLUSH, &not_used)))
 
1592
  if (! ((*lock)= mysql_lock_tables(session, &table, 1,
 
1593
                                    DRIZZLE_LOCK_IGNORE_FLUSH, &not_used)))
1643
1594
  {
1644
1595
    if (*lock)
1645
1596
    {
1646
 
      session->unlockTables(*lock);
 
1597
      mysql_unlock_tables(session, *lock);
1647
1598
      *lock= 0;
1648
1599
    }
1649
1600
 
1650
 
    if (not create_info->table_existed)
1651
 
      session->drop_open_table(table, identifier);
 
1601
    if (!create_info->table_existed)
 
1602
      session->drop_open_table(table, create_table->db, create_table->table_name);
1652
1603
    return NULL;
1653
1604
  }
1654
1605
 
1659
1610
int
1660
1611
select_create::prepare(List<Item> &values, Select_Lex_Unit *u)
1661
1612
{
1662
 
  DrizzleLock *extra_lock= NULL;
 
1613
  bool lex_identified_temp_table= (table_proto->type() == message::Table::TEMPORARY);
 
1614
 
 
1615
  DRIZZLE_LOCK *extra_lock= NULL;
1663
1616
  /*
1664
 
    For replication, the CREATE-SELECT statement is written
1665
 
    in two pieces: the first transaction messsage contains 
1666
 
    the CREATE TABLE statement as a CreateTableStatement message
1667
 
    necessary to create the table.
1668
 
    
1669
 
    The second transaction message contains all the InsertStatement
1670
 
    and associated InsertRecords that should go into the table.
 
1617
    For row-based replication, the CREATE-SELECT statement is written
 
1618
    in two pieces: the first one contain the CREATE TABLE statement
 
1619
    necessary to create the table and the second part contain the rows
 
1620
    that should go into the table.
 
1621
 
 
1622
    For non-temporary tables, the start of the CREATE-SELECT
 
1623
    implicitly commits the previous transaction, and all events
 
1624
    forming the statement will be stored the transaction cache. At end
 
1625
    of the statement, the entire statement is committed as a
 
1626
    transaction, and all events are written to the binary log.
 
1627
 
 
1628
    On the master, the table is locked for the duration of the
 
1629
    statement, but since the CREATE part is replicated as a simple
 
1630
    statement, there is no way to lock the table for accesses on the
 
1631
    slave.  Hence, we have to hold on to the CREATE part of the
 
1632
    statement until the statement has finished.
1671
1633
   */
1672
1634
 
1673
1635
  unit= u;
1674
1636
 
1675
 
  if (not (table= create_table_from_items(session, create_info, create_table,
1676
 
                                          table_proto,
1677
 
                                          alter_info, &values,
1678
 
                                          is_if_not_exists,
1679
 
                                          &extra_lock, identifier)))
1680
 
  {
 
1637
  /*
 
1638
    Start a statement transaction before the create if we are using
 
1639
    row-based replication for the statement.  If we are creating a
 
1640
    temporary table, we need to start a statement transaction.
 
1641
  */
 
1642
 
 
1643
  if (!(table= create_table_from_items(session, create_info, create_table,
 
1644
                                       table_proto,
 
1645
                                       alter_info, &values,
 
1646
                                       is_if_not_exists,
 
1647
                                       &extra_lock)))
1681
1648
    return(-1);                         // abort() deletes table
1682
 
  }
1683
1649
 
1684
1650
  if (extra_lock)
1685
1651
  {
1686
1652
    assert(m_plock == NULL);
1687
1653
 
1688
 
    if (identifier.isTmp())
 
1654
    if (lex_identified_temp_table)
1689
1655
      m_plock= &m_lock;
1690
1656
    else
1691
1657
      m_plock= &session->extra_lock;
1693
1659
    *m_plock= extra_lock;
1694
1660
  }
1695
1661
 
1696
 
  if (table->getShare()->sizeFields() < values.elements)
 
1662
  if (table->s->fields < values.elements)
1697
1663
  {
1698
1664
    my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1);
1699
1665
    return(-1);
1700
1666
  }
1701
1667
 
1702
1668
 /* First field to copy */
1703
 
  field= table->getFields() + table->getShare()->sizeFields() - values.elements;
 
1669
  field= table->field+table->s->fields - values.elements;
1704
1670
 
1705
1671
  /* Mark all fields that are given values */
1706
1672
  for (Field **f= field ; *f ; f++)
1707
 
  {
1708
 
    table->setWriteSet((*f)->position());
1709
 
  }
 
1673
    table->setWriteSet((*f)->field_index);
1710
1674
 
1711
1675
  /* Don't set timestamp if used */
1712
1676
  table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
1737
1701
 
1738
1702
void select_create::send_error(uint32_t errcode,const char *err)
1739
1703
{
 
1704
 
 
1705
 
1740
1706
  /*
1741
1707
    This will execute any rollbacks that are necessary before writing
1742
1708
    the transcation cache.
1766
1732
      tables.  This can fail, but we should unlock the table
1767
1733
      nevertheless.
1768
1734
    */
1769
 
    if (!table->getShare()->getType())
 
1735
    if (!table->s->tmp_table)
1770
1736
    {
1771
1737
      TransactionServices &transaction_services= TransactionServices::singleton();
1772
 
      transaction_services.autocommitOrRollback(session, 0);
 
1738
      transaction_services.ha_autocommit_or_rollback(session, 0);
1773
1739
      (void) session->endActiveTransaction();
1774
1740
    }
1775
1741
 
1777
1743
    table->cursor->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
1778
1744
    if (m_plock)
1779
1745
    {
1780
 
      session->unlockTables(*m_plock);
 
1746
      mysql_unlock_tables(session, *m_plock);
1781
1747
      *m_plock= NULL;
1782
1748
      m_plock= NULL;
1783
1749
    }
1788
1754
 
1789
1755
void select_create::abort()
1790
1756
{
 
1757
 
 
1758
 
1791
1759
  /*
1792
1760
    In select_insert::abort() we roll back the statement, including
1793
1761
    truncating the transaction cache of the binary log. To do this, we
1804
1772
    log state.
1805
1773
  */
1806
1774
  select_insert::abort();
 
1775
  session->transaction.stmt.modified_non_trans_table= false;
 
1776
 
1807
1777
 
1808
1778
  if (m_plock)
1809
1779
  {
1810
 
    session->unlockTables(*m_plock);
 
1780
    mysql_unlock_tables(session, *m_plock);
1811
1781
    *m_plock= NULL;
1812
1782
    m_plock= NULL;
1813
1783
  }
1816
1786
  {
1817
1787
    table->cursor->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
1818
1788
    table->cursor->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
1819
 
    if (not create_info->table_existed)
1820
 
      session->drop_open_table(table, identifier);
 
1789
    if (!create_info->table_existed)
 
1790
      session->drop_open_table(table, create_table->db, create_table->table_name);
1821
1791
    table= NULL;                                    // Safety
1822
1792
  }
1823
1793
}