← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/statement/alter_table.cc

  • Committer: Brian Aker
  • Date: 2010-12-08 22:35:56 UTC
  • mfrom: (1819.9.158 update-innobase)
  • Revision ID: brian@tangent.org-20101208223556-37mi4omqg7lkjzf3
Merge in Stewart's changes, 1.3 changes.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/statement/alter_table.cc
 
1
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
 
2
 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 
3
 *
 
4
 *  Copyright (C) 2009 Sun Microsystems
 
5
 *
 
6
 *  This program is free software; you can redistribute it and/or modify
 
7
 *  it under the terms of the GNU General Public License as published by
 
8
 *  the Free Software Foundation; either version 2 of the License, or
 
9
 *  (at your option) any later version.
 
10
 *
 
11
 *  This program is distributed in the hope that it will be useful,
 
12
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
 *  GNU General Public License for more details.
 
15
 *
 
16
 *  You should have received a copy of the GNU General Public License
 
17
 *  along with this program; if not, write to the Free Software
 
18
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 
19
 */
 
20
 
 
21
#include "config.h"
 
22
 
 
23
#include <fcntl.h>
 
24
 
 
25
#include <sstream>
 
26
 
 
27
#include "drizzled/show.h"
 
28
#include "drizzled/lock.h"
 
29
#include "drizzled/session.h"
 
30
#include "drizzled/statement/alter_table.h"
 
31
#include "drizzled/global_charset_info.h"
 
32
 
 
33
 
 
34
#include "drizzled/gettext.h"
 
35
#include "drizzled/data_home.h"
 
36
#include "drizzled/sql_table.h"
 
37
#include "drizzled/table_proto.h"
 
38
#include "drizzled/optimizer/range.h"
 
39
#include "drizzled/time_functions.h"
 
40
#include "drizzled/records.h"
 
41
#include "drizzled/pthread_globals.h"
 
42
#include "drizzled/internal/my_sys.h"
 
43
#include "drizzled/internal/iocache.h"
 
44
 
 
45
#include "drizzled/transaction_services.h"
 
46
 
 
47
#include "drizzled/filesort.h"
 
48
 
 
49
#include "drizzled/message.h"
 
50
 
 
51
using namespace std;
 
52
 
 
53
namespace drizzled
 
54
{
 
55
 
 
56
extern pid_t current_pid;
 
57
 
 
58
static int copy_data_between_tables(Session *session,
 
59
                                    Table *from,Table *to,
 
60
                                    List<CreateField> &create,
 
61
                                    bool ignore,
 
62
                                    uint32_t order_num,
 
63
                                    Order *order,
 
64
                                    ha_rows *copied,
 
65
                                    ha_rows *deleted,
 
66
                                    enum enum_enable_or_disable keys_onoff,
 
67
                                    bool error_if_not_empty);
 
68
 
 
69
static bool mysql_prepare_alter_table(Session *session,
 
70
                                      Table *table,
 
71
                                      HA_CREATE_INFO *create_info,
 
72
                                      const message::Table &original_proto,
 
73
                                      message::Table &table_message,
 
74
                                      AlterInfo *alter_info);
 
75
 
 
76
static int create_temporary_table(Session *session,
 
77
                                  TableIdentifier &identifier,
 
78
                                  HA_CREATE_INFO *create_info,
 
79
                                  message::Table &create_message,
 
80
                                  AlterInfo *alter_info);
 
81
 
 
82
static Table *open_alter_table(Session *session, Table *table, TableIdentifier &identifier);
 
83
 
 
84
bool statement::AlterTable::execute()
 
85
{
 
86
  TableList *first_table= (TableList *) session->lex->select_lex.table_list.first;
 
87
  TableList *all_tables= session->lex->query_tables;
 
88
  assert(first_table == all_tables && first_table != 0);
 
89
  Select_Lex *select_lex= &session->lex->select_lex;
 
90
  bool need_start_waiting= false;
 
91
 
 
92
  if (is_engine_set)
 
93
  {
 
94
    create_info.db_type= 
 
95
      plugin::StorageEngine::findByName(*session, create_table_message.engine().name());
 
96
 
 
97
    if (create_info.db_type == NULL)
 
98
    {
 
99
      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), 
 
100
               create_table_message.engine().name().c_str());
 
101
 
 
102
      return true;
 
103
    }
 
104
  }
 
105
 
 
106
  /* Must be set in the parser */
 
107
  assert(select_lex->db);
 
108
 
 
109
  /* Chicken/Egg... we need to search for the table, to know if the table exists, so we can build a full identifier from it */
 
110
  message::table::shared_ptr original_table_message;
 
111
  {
 
112
    TableIdentifier identifier(first_table->getSchemaName(), first_table->getTableName());
 
113
    if (plugin::StorageEngine::getTableDefinition(*session, identifier, original_table_message) != EEXIST)
 
114
    {
 
115
      std::string path;
 
116
      identifier.getSQLPath(path);
 
117
      my_error(ER_BAD_TABLE_ERROR, MYF(0), path.c_str());
 
118
      return true;
 
119
    }
 
120
 
 
121
    if (not  create_info.db_type)
 
122
    {
 
123
      create_info.db_type= 
 
124
        plugin::StorageEngine::findByName(*session, original_table_message->engine().name());
 
125
 
 
126
      if (not create_info.db_type)
 
127
      {
 
128
        std::string path;
 
129
        identifier.getSQLPath(path);
 
130
        my_error(ER_BAD_TABLE_ERROR, MYF(0), path.c_str());
 
131
        return true;
 
132
      }
 
133
    }
 
134
  }
 
135
 
 
136
  if (not validateCreateTableOption())
 
137
  {
 
138
    return true;
 
139
  }
 
140
 
 
141
  /* ALTER TABLE ends previous transaction */
 
142
  if (not session->endActiveTransaction())
 
143
  {
 
144
    return true;
 
145
  }
 
146
 
 
147
  if (not (need_start_waiting= not session->wait_if_global_read_lock(0, 1)))
 
148
  {
 
149
    return true;
 
150
  }
 
151
 
 
152
  bool res;
 
153
  if (original_table_message->type() == message::Table::STANDARD )
 
154
  {
 
155
    TableIdentifier identifier(first_table->getSchemaName(), first_table->getTableName());
 
156
    TableIdentifier new_identifier(select_lex->db ? select_lex->db : first_table->getSchemaName(),
 
157
                                   session->lex->name.str ? session->lex->name.str : first_table->getTableName());
 
158
 
 
159
    res= alter_table(session, 
 
160
                     identifier,
 
161
                     new_identifier,
 
162
                     &create_info,
 
163
                     *original_table_message,
 
164
                     create_table_message,
 
165
                     first_table,
 
166
                     &alter_info,
 
167
                     select_lex->order_list.elements,
 
168
                     (Order *) select_lex->order_list.first,
 
169
                     session->lex->ignore);
 
170
  }
 
171
  else
 
172
  {
 
173
    TableIdentifier catch22(first_table->getSchemaName(), first_table->getTableName());
 
174
    Table *table= session->find_temporary_table(catch22);
 
175
    assert(table);
 
176
    {
 
177
      TableIdentifier identifier(first_table->getSchemaName(), first_table->getTableName(), table->getMutableShare()->getPath());
 
178
      TableIdentifier new_identifier(select_lex->db ? select_lex->db : first_table->getSchemaName(),
 
179
                                     session->lex->name.str ? session->lex->name.str : first_table->getTableName(),
 
180
                                     table->getMutableShare()->getPath());
 
181
 
 
182
      res= alter_table(session, 
 
183
                       identifier,
 
184
                       new_identifier,
 
185
                       &create_info,
 
186
                       *original_table_message,
 
187
                       create_table_message,
 
188
                       first_table,
 
189
                       &alter_info,
 
190
                       select_lex->order_list.elements,
 
191
                       (Order *) select_lex->order_list.first,
 
192
                       session->lex->ignore);
 
193
    }
 
194
  }
 
195
 
 
196
  /*
 
197
     Release the protection against the global read lock and wake
 
198
     everyone, who might want to set a global read lock.
 
199
   */
 
200
  session->startWaitingGlobalReadLock();
 
201
 
 
202
  return res;
 
203
}
 
204
 
 
205
 
 
206
/**
 
207
  Prepare column and key definitions for CREATE TABLE in ALTER Table.
 
208
 
 
209
  This function transforms parse output of ALTER Table - lists of
 
210
  columns and keys to add, drop or modify into, essentially,
 
211
  CREATE TABLE definition - a list of columns and keys of the new
 
212
  table. While doing so, it also performs some (bug not all)
 
213
  semantic checks.
 
214
 
 
215
  This function is invoked when we know that we're going to
 
216
  perform ALTER Table via a temporary table -- i.e. fast ALTER Table
 
217
  is not possible, perhaps because the ALTER statement contains
 
218
  instructions that require change in table data, not only in
 
219
  table definition or indexes.
 
220
 
 
221
  @param[in,out]  session         thread handle. Used as a memory pool
 
222
                              and source of environment information.
 
223
  @param[in]      table       the source table, open and locked
 
224
                              Used as an interface to the storage engine
 
225
                              to acquire additional information about
 
226
                              the original table.
 
227
  @param[in,out]  create_info A blob with CREATE/ALTER Table
 
228
                              parameters
 
229
  @param[in,out]  alter_info  Another blob with ALTER/CREATE parameters.
 
230
                              Originally create_info was used only in
 
231
                              CREATE TABLE and alter_info only in ALTER Table.
 
232
                              But since ALTER might end-up doing CREATE,
 
233
                              this distinction is gone and we just carry
 
234
                              around two structures.
 
235
 
 
236
  @return
 
237
    Fills various create_info members based on information retrieved
 
238
    from the storage engine.
 
239
    Sets create_info->varchar if the table has a VARCHAR column.
 
240
    Prepares alter_info->create_list and alter_info->key_list with
 
241
    columns and keys of the new table.
 
242
  @retval true   error, out of memory or a semantical error in ALTER
 
243
                 Table instructions
 
244
  @retval false  success
 
245
*/
 
246
static bool mysql_prepare_alter_table(Session *session,
 
247
                                      Table *table,
 
248
                                      HA_CREATE_INFO *create_info,
 
249
                                      const message::Table &original_proto,
 
250
                                      message::Table &table_message,
 
251
                                      AlterInfo *alter_info)
 
252
{
 
253
  /* New column definitions are added here */
 
254
  List<CreateField> new_create_list;
 
255
  /* New key definitions are added here */
 
256
  List<Key> new_key_list;
 
257
  List_iterator<AlterDrop> drop_it(alter_info->drop_list);
 
258
  List_iterator<CreateField> def_it(alter_info->create_list);
 
259
  List_iterator<AlterColumn> alter_it(alter_info->alter_list);
 
260
  List_iterator<Key> key_it(alter_info->key_list);
 
261
  List_iterator<CreateField> find_it(new_create_list);
 
262
  List_iterator<CreateField> field_it(new_create_list);
 
263
  List<Key_part_spec> key_parts;
 
264
  uint32_t used_fields= create_info->used_fields;
 
265
  KeyInfo *key_info= table->key_info;
 
266
  bool rc= true;
 
267
 
 
268
  /* Let new create options override the old ones */
 
269
  message::Table::TableOptions *table_options;
 
270
  table_options= table_message.mutable_options();
 
271
 
 
272
  if (! (used_fields & HA_CREATE_USED_DEFAULT_CHARSET))
 
273
    create_info->default_table_charset= table->getShare()->table_charset;
 
274
  if (! (used_fields & HA_CREATE_USED_AUTO) &&
 
275
      table->found_next_number_field)
 
276
  {
 
277
    /* Table has an autoincrement, copy value to new table */
 
278
    table->cursor->info(HA_STATUS_AUTO);
 
279
    create_info->auto_increment_value= table->cursor->stats.auto_increment_value;
 
280
    if (create_info->auto_increment_value != original_proto.options().auto_increment_value())
 
281
      table_options->set_has_user_set_auto_increment_value(false);
 
282
  }
 
283
  table->restoreRecordAsDefault(); /* Empty record for DEFAULT */
 
284
  CreateField *def;
 
285
 
 
286
  /* First collect all fields from table which isn't in drop_list */
 
287
  Field **f_ptr;
 
288
  Field *field;
 
289
  for (f_ptr= table->getFields(); (field= *f_ptr); f_ptr++)
 
290
  {
 
291
    /* Check if field should be dropped */
 
292
    AlterDrop *drop;
 
293
    drop_it.rewind();
 
294
    while ((drop= drop_it++))
 
295
    {
 
296
      if (drop->type == AlterDrop::COLUMN &&
 
297
          ! my_strcasecmp(system_charset_info, field->field_name, drop->name))
 
298
      {
 
299
        /* Reset auto_increment value if it was dropped */
 
300
        if (MTYP_TYPENR(field->unireg_check) == Field::NEXT_NUMBER &&
 
301
            ! (used_fields & HA_CREATE_USED_AUTO))
 
302
        {
 
303
          create_info->auto_increment_value= 0;
 
304
          create_info->used_fields|= HA_CREATE_USED_AUTO;
 
305
        }
 
306
        break;
 
307
      }
 
308
    }
 
309
    if (drop)
 
310
    {
 
311
      drop_it.remove();
 
312
      continue;
 
313
    }
 
314
    
 
315
    /* Mark that we will read the field */
 
316
    field->setReadSet();
 
317
 
 
318
    /* Check if field is changed */
 
319
    def_it.rewind();
 
320
    while ((def= def_it++))
 
321
    {
 
322
      if (def->change &&
 
323
          ! my_strcasecmp(system_charset_info, field->field_name, def->change))
 
324
              break;
 
325
    }
 
326
    if (def)
 
327
    {
 
328
      /* Field is changed */
 
329
      def->field= field;
 
330
      if (! def->after)
 
331
      {
 
332
        new_create_list.push_back(def);
 
333
        def_it.remove();
 
334
      }
 
335
    }
 
336
    else
 
337
    {
 
338
      /*
 
339
        This field was not dropped and not changed, add it to the list
 
340
        for the new table.
 
341
      */
 
342
      def= new CreateField(field, field);
 
343
      new_create_list.push_back(def);
 
344
      alter_it.rewind(); /* Change default if ALTER */
 
345
      AlterColumn *alter;
 
346
      while ((alter= alter_it++))
 
347
      {
 
348
        if (! my_strcasecmp(system_charset_info,field->field_name, alter->name))
 
349
          break;
 
350
      }
 
351
      if (alter)
 
352
      {
 
353
        if (def->sql_type == DRIZZLE_TYPE_BLOB)
 
354
        {
 
355
          my_error(ER_BLOB_CANT_HAVE_DEFAULT, MYF(0), def->change);
 
356
          return true;
 
357
        }
 
358
        if ((def->def= alter->def))
 
359
        {
 
360
          /* Use new default */
 
361
          def->flags&= ~NO_DEFAULT_VALUE_FLAG;
 
362
        }
 
363
        else
 
364
        {
 
365
          def->flags|= NO_DEFAULT_VALUE_FLAG;
 
366
        }
 
367
        alter_it.remove();
 
368
      }
 
369
    }
 
370
  }
 
371
  def_it.rewind();
 
372
  while ((def= def_it++)) /* Add new columns */
 
373
  {
 
374
    if (def->change && ! def->field)
 
375
    {
 
376
      my_error(ER_BAD_FIELD_ERROR, MYF(0), def->change, table->getMutableShare()->getTableName());
 
377
      return true;
 
378
    }
 
379
    /*
 
380
      If we have been given a field which has no default value, and is not null then we need to bail.
 
381
    */
 
382
    if (not (~def->flags & (NO_DEFAULT_VALUE_FLAG | NOT_NULL_FLAG)) and not def->change)
 
383
    {
 
384
      alter_info->error_if_not_empty= true;
 
385
    }
 
386
    if (! def->after)
 
387
    {
 
388
      new_create_list.push_back(def);
 
389
    }
 
390
    else if (def->after == first_keyword)
 
391
    {
 
392
      new_create_list.push_front(def);
 
393
    }
 
394
    else
 
395
    {
 
396
      CreateField *find;
 
397
      find_it.rewind();
 
398
      while ((find= find_it++)) /* Add new columns */
 
399
      {
 
400
        if (! my_strcasecmp(system_charset_info,def->after, find->field_name))
 
401
          break;
 
402
      }
 
403
      if (! find)
 
404
      {
 
405
        my_error(ER_BAD_FIELD_ERROR, MYF(0), def->after, table->getMutableShare()->getTableName());
 
406
        return true;
 
407
      }
 
408
      find_it.after(def); /* Put element after this */
 
409
      /*
 
410
        XXX: hack for Bug#28427.
 
411
        If column order has changed, force OFFLINE ALTER Table
 
412
        without querying engine capabilities.  If we ever have an
 
413
        engine that supports online ALTER Table CHANGE COLUMN
 
414
        <name> AFTER <name1> (Falcon?), this fix will effectively
 
415
        disable the capability.
 
416
        TODO: detect the situation in compare_tables, behave based
 
417
        on engine capabilities.
 
418
      */
 
419
      if (alter_info->build_method == HA_BUILD_ONLINE)
 
420
      {
 
421
        my_error(ER_NOT_SUPPORTED_YET, MYF(0), session->getQueryString()->c_str());
 
422
        return true;
 
423
      }
 
424
      alter_info->build_method= HA_BUILD_OFFLINE;
 
425
    }
 
426
  }
 
427
  if (alter_info->alter_list.elements)
 
428
  {
 
429
    my_error(ER_BAD_FIELD_ERROR,
 
430
             MYF(0),
 
431
             alter_info->alter_list.head()->name,
 
432
             table->getMutableShare()->getTableName());
 
433
    return true;
 
434
  }
 
435
  if (! new_create_list.elements)
 
436
  {
 
437
    my_message(ER_CANT_REMOVE_ALL_FIELDS,
 
438
               ER(ER_CANT_REMOVE_ALL_FIELDS),
 
439
               MYF(0));
 
440
    return true;
 
441
  }
 
442
 
 
443
  /*
 
444
    Collect all keys which isn't in drop list. Add only those
 
445
    for which some fields exists.
 
446
  */
 
447
  for (uint32_t i= 0; i < table->getShare()->sizeKeys(); i++, key_info++)
 
448
  {
 
449
    char *key_name= key_info->name;
 
450
    AlterDrop *drop;
 
451
    drop_it.rewind();
 
452
    while ((drop= drop_it++))
 
453
    {
 
454
      if (drop->type == AlterDrop::KEY &&
 
455
          ! my_strcasecmp(system_charset_info, key_name, drop->name))
 
456
        break;
 
457
    }
 
458
    if (drop)
 
459
    {
 
460
      drop_it.remove();
 
461
      continue;
 
462
    }
 
463
 
 
464
    KeyPartInfo *key_part= key_info->key_part;
 
465
    key_parts.empty();
 
466
    for (uint32_t j= 0; j < key_info->key_parts; j++, key_part++)
 
467
    {
 
468
      if (! key_part->field)
 
469
              continue; /* Wrong field (from UNIREG) */
 
470
 
 
471
      const char *key_part_name= key_part->field->field_name;
 
472
      CreateField *cfield;
 
473
      field_it.rewind();
 
474
      while ((cfield= field_it++))
 
475
      {
 
476
        if (cfield->change)
 
477
        {
 
478
          if (! my_strcasecmp(system_charset_info, key_part_name, cfield->change))
 
479
            break;
 
480
        }
 
481
        else if (! my_strcasecmp(system_charset_info, key_part_name, cfield->field_name))
 
482
          break;
 
483
      }
 
484
      if (! cfield)
 
485
              continue; /* Field is removed */
 
486
      
 
487
      uint32_t key_part_length= key_part->length;
 
488
      if (cfield->field) /* Not new field */
 
489
      {
 
490
        /*
 
491
          If the field can't have only a part used in a key according to its
 
492
          new type, or should not be used partially according to its
 
493
          previous type, or the field length is less than the key part
 
494
          length, unset the key part length.
 
495
 
 
496
          We also unset the key part length if it is the same as the
 
497
          old field's length, so the whole new field will be used.
 
498
 
 
499
          BLOBs may have cfield->length == 0, which is why we test it before
 
500
          checking whether cfield->length < key_part_length (in chars).
 
501
         */
 
502
        if (! Field::type_can_have_key_part(cfield->field->type()) ||
 
503
            ! Field::type_can_have_key_part(cfield->sql_type) ||
 
504
            (cfield->field->field_length == key_part_length) ||
 
505
            (cfield->length &&
 
506
             (cfield->length < key_part_length / key_part->field->charset()->mbmaxlen)))
 
507
          key_part_length= 0; /* Use whole field */
 
508
      }
 
509
      key_part_length/= key_part->field->charset()->mbmaxlen;
 
510
      key_parts.push_back(new Key_part_spec(cfield->field_name,
 
511
                                            strlen(cfield->field_name),
 
512
                                            key_part_length));
 
513
    }
 
514
    if (key_parts.elements)
 
515
    {
 
516
      KEY_CREATE_INFO key_create_info;
 
517
      Key *key;
 
518
      enum Key::Keytype key_type;
 
519
      memset(&key_create_info, 0, sizeof(key_create_info));
 
520
 
 
521
      key_create_info.algorithm= key_info->algorithm;
 
522
      if (key_info->flags & HA_USES_BLOCK_SIZE)
 
523
        key_create_info.block_size= key_info->block_size;
 
524
      if (key_info->flags & HA_USES_COMMENT)
 
525
        key_create_info.comment= key_info->comment;
 
526
 
 
527
      if (key_info->flags & HA_NOSAME)
 
528
      {
 
529
        if (is_primary_key_name(key_name))
 
530
          key_type= Key::PRIMARY;
 
531
        else
 
532
          key_type= Key::UNIQUE;
 
533
      }
 
534
      else
 
535
        key_type= Key::MULTIPLE;
 
536
 
 
537
      key= new Key(key_type,
 
538
                   key_name,
 
539
                   strlen(key_name),
 
540
                   &key_create_info,
 
541
                   test(key_info->flags & HA_GENERATED_KEY),
 
542
                   key_parts);
 
543
      new_key_list.push_back(key);
 
544
    }
 
545
  }
 
546
 
 
547
  /* Copy over existing foreign keys */
 
548
  for (int j= 0; j < original_proto.fk_constraint_size(); j++)
 
549
  {
 
550
    AlterDrop *drop;
 
551
    drop_it.rewind();
 
552
    while ((drop= drop_it++))
 
553
    {
 
554
      if (drop->type == AlterDrop::FOREIGN_KEY &&
 
555
          ! my_strcasecmp(system_charset_info, original_proto.fk_constraint(j).name().c_str(), drop->name))
 
556
      {
 
557
        break;
 
558
      }
 
559
    }
 
560
    if (drop)
 
561
    {
 
562
      drop_it.remove();
 
563
      continue;
 
564
    }
 
565
 
 
566
    message::Table::ForeignKeyConstraint *pfkey= table_message.add_fk_constraint();
 
567
    *pfkey= original_proto.fk_constraint(j);
 
568
  }
 
569
 
 
570
  {
 
571
    Key *key;
 
572
    while ((key= key_it++)) /* Add new keys */
 
573
    {
 
574
      if (key->type == Key::FOREIGN_KEY)
 
575
      {
 
576
        if (((Foreign_key *)key)->validate(new_create_list))
 
577
        {
 
578
          return true;
 
579
        }
 
580
 
 
581
        Foreign_key *fkey= (Foreign_key*)key;
 
582
        add_foreign_key_to_table_message(&table_message,
 
583
                                         fkey->name.str,
 
584
                                         fkey->columns,
 
585
                                         fkey->ref_table,
 
586
                                         fkey->ref_columns,
 
587
                                         fkey->delete_opt,
 
588
                                         fkey->update_opt,
 
589
                                         fkey->match_opt);
 
590
      }
 
591
 
 
592
      if (key->type != Key::FOREIGN_KEY)
 
593
        new_key_list.push_back(key);
 
594
 
 
595
      if (key->name.str && is_primary_key_name(key->name.str))
 
596
      {
 
597
        my_error(ER_WRONG_NAME_FOR_INDEX,
 
598
                 MYF(0),
 
599
                 key->name.str);
 
600
        return true;
 
601
      }
 
602
    }
 
603
  }
 
604
 
 
605
  /* Fix names of foreign keys being added */
 
606
  for (int j= 0; j < table_message.fk_constraint_size(); j++)
 
607
  {
 
608
    if (! table_message.fk_constraint(j).has_name())
 
609
    {
 
610
      std::string name(table->getMutableShare()->getTableName());
 
611
      char number[20];
 
612
 
 
613
      name.append("_ibfk_");
 
614
      snprintf(number, sizeof(number), "%d", j+1);
 
615
      name.append(number);
 
616
 
 
617
      message::Table::ForeignKeyConstraint *pfkey= table_message.mutable_fk_constraint(j);
 
618
      pfkey->set_name(name);
 
619
    }
 
620
  }
 
621
 
 
622
  if (alter_info->drop_list.elements)
 
623
  {
 
624
    my_error(ER_CANT_DROP_FIELD_OR_KEY,
 
625
             MYF(0),
 
626
             alter_info->drop_list.head()->name);
 
627
    return true;
 
628
  }
 
629
  if (alter_info->alter_list.elements)
 
630
  {
 
631
    my_error(ER_CANT_DROP_FIELD_OR_KEY,
 
632
             MYF(0),
 
633
             alter_info->alter_list.head()->name);
 
634
    return true;
 
635
  }
 
636
 
 
637
  if (not table_message.options().has_comment()
 
638
      && table->getMutableShare()->hasComment())
 
639
    table_options->set_comment(table->getMutableShare()->getComment());
 
640
 
 
641
  if (table->getShare()->getType())
 
642
  {
 
643
    table_message.set_type(message::Table::TEMPORARY);
 
644
  }
 
645
 
 
646
  table_message.set_creation_timestamp(table->getShare()->getTableProto()->creation_timestamp());
 
647
  table_message.set_version(table->getShare()->getTableProto()->version());
 
648
  table_message.set_uuid(table->getShare()->getTableProto()->uuid());
 
649
 
 
650
  rc= false;
 
651
  alter_info->create_list.swap(new_create_list);
 
652
  alter_info->key_list.swap(new_key_list);
 
653
 
 
654
  size_t num_engine_options= table_message.engine().options_size();
 
655
  size_t original_num_engine_options= original_proto.engine().options_size();
 
656
  for (size_t x= 0; x < original_num_engine_options; ++x)
 
657
  {
 
658
    bool found= false;
 
659
 
 
660
    for (size_t y= 0; y < num_engine_options; ++y)
 
661
    {
 
662
      found= not table_message.engine().options(y).name().compare(original_proto.engine().options(x).name());
 
663
      
 
664
      if (found)
 
665
        break;
 
666
    }
 
667
 
 
668
    if (not found)
 
669
    {
 
670
      message::Engine::Option *opt= table_message.mutable_engine()->add_options();
 
671
 
 
672
      opt->set_name(original_proto.engine().options(x).name());
 
673
      opt->set_state(original_proto.engine().options(x).state());
 
674
    }
 
675
  }
 
676
 
 
677
  drizzled::message::update(table_message);
 
678
 
 
679
  return false;
 
680
}
 
681
 
 
682
/* table_list should contain just one table */
 
683
static int mysql_discard_or_import_tablespace(Session *session,
 
684
                                              TableList *table_list,
 
685
                                              enum tablespace_op_type tablespace_op)
 
686
{
 
687
  Table *table;
 
688
  bool discard;
 
689
  int error;
 
690
 
 
691
  /*
 
692
    Note that DISCARD/IMPORT TABLESPACE always is the only operation in an
 
693
    ALTER Table
 
694
  */
 
695
 
 
696
  TransactionServices &transaction_services= TransactionServices::singleton();
 
697
  session->set_proc_info("discard_or_import_tablespace");
 
698
 
 
699
  discard= test(tablespace_op == DISCARD_TABLESPACE);
 
700
 
 
701
 /*
 
702
   We set this flag so that ha_innobase::open and ::external_lock() do
 
703
   not complain when we lock the table
 
704
 */
 
705
  session->tablespace_op= true;
 
706
  if (!(table= session->openTableLock(table_list, TL_WRITE)))
 
707
  {
 
708
    session->tablespace_op= false;
 
709
    return -1;
 
710
  }
 
711
 
 
712
  error= table->cursor->ha_discard_or_import_tablespace(discard);
 
713
 
 
714
  session->set_proc_info("end");
 
715
 
 
716
  if (error)
 
717
    goto err;
 
718
 
 
719
  /* The ALTER Table is always in its own transaction */
 
720
  error= transaction_services.autocommitOrRollback(session, false);
 
721
  if (not session->endActiveTransaction())
 
722
    error=1;
 
723
 
 
724
  if (error)
 
725
    goto err;
 
726
 
 
727
  write_bin_log(session, *session->getQueryString());
 
728
 
 
729
err:
 
730
  (void) transaction_services.autocommitOrRollback(session, error);
 
731
  session->tablespace_op=false;
 
732
 
 
733
  if (error == 0)
 
734
  {
 
735
    session->my_ok();
 
736
    return 0;
 
737
  }
 
738
 
 
739
  table->print_error(error, MYF(0));
 
740
 
 
741
  return -1;
 
742
}
 
743
 
 
744
/**
 
745
  Manages enabling/disabling of indexes for ALTER Table
 
746
 
 
747
  SYNOPSIS
 
748
    alter_table_manage_keys()
 
749
      table                  Target table
 
750
      indexes_were_disabled  Whether the indexes of the from table
 
751
                             were disabled
 
752
      keys_onoff             ENABLE | DISABLE | LEAVE_AS_IS
 
753
 
 
754
  RETURN VALUES
 
755
    false  OK
 
756
    true   Error
 
757
*/
 
758
static bool alter_table_manage_keys(Session *session,
 
759
                                    Table *table, int indexes_were_disabled,
 
760
                                    enum enum_enable_or_disable keys_onoff)
 
761
{
 
762
  int error= 0;
 
763
  switch (keys_onoff) {
 
764
  case ENABLE:
 
765
    error= table->cursor->ha_enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
 
766
    break;
 
767
  case LEAVE_AS_IS:
 
768
    if (not indexes_were_disabled)
 
769
      break;
 
770
    /* fall-through: disabled indexes */
 
771
  case DISABLE:
 
772
    error= table->cursor->ha_disable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
 
773
  }
 
774
 
 
775
  if (error == HA_ERR_WRONG_COMMAND)
 
776
  {
 
777
    push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
778
                        ER_ILLEGAL_HA, ER(ER_ILLEGAL_HA),
 
779
                        table->getMutableShare()->getTableName());
 
780
    error= 0;
 
781
  } else if (error)
 
782
    table->print_error(error, MYF(0));
 
783
 
 
784
  return(error);
 
785
}
 
786
 
 
787
static bool lockTableIfDifferent(Session &session,
 
788
                                 TableIdentifier &original_table_identifier,
 
789
                                 TableIdentifier &new_table_identifier,
 
790
                                 Table *name_lock)
 
791
{
 
792
  /* Check that we are not trying to rename to an existing table */
 
793
  if (not (original_table_identifier == new_table_identifier))
 
794
  {
 
795
    if (original_table_identifier.isTmp())
 
796
    {
 
797
 
 
798
      if (session.find_temporary_table(new_table_identifier))
 
799
      {
 
800
        std::string path;
 
801
        new_table_identifier.getSQLPath(path);
 
802
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), path.c_str());
 
803
        return false;
 
804
      }
 
805
    }
 
806
    else
 
807
    {
 
808
      if (session.lock_table_name_if_not_cached(new_table_identifier, &name_lock))
 
809
      {
 
810
        return false;
 
811
      }
 
812
 
 
813
      if (not name_lock)
 
814
      {
 
815
        std::string path;
 
816
        new_table_identifier.getSQLPath(path);
 
817
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), path.c_str());
 
818
        return false;
 
819
      }
 
820
 
 
821
      if (plugin::StorageEngine::doesTableExist(session, new_table_identifier))
 
822
      {
 
823
        std::string path;
 
824
        new_table_identifier.getSQLPath(path);
 
825
 
 
826
        /* Table will be closed by Session::executeCommand() */
 
827
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), path.c_str());
 
828
 
 
829
        table::Cache::singleton().mutex().lock(); /* ALTER TABLe */
 
830
        session.unlink_open_table(name_lock);
 
831
        table::Cache::singleton().mutex().unlock();
 
832
 
 
833
        return false;
 
834
      }
 
835
    }
 
836
  }
 
837
 
 
838
  return true;
 
839
}
 
840
 
 
841
/**
 
842
  Alter table
 
843
 
 
844
  SYNOPSIS
 
845
    alter_table()
 
846
      session              Thread handle
 
847
      new_db           If there is a RENAME clause
 
848
      new_name         If there is a RENAME clause
 
849
      create_info      Information from the parsing phase about new
 
850
                       table properties.
 
851
      table_list       The table to change.
 
852
      alter_info       Lists of fields, keys to be changed, added
 
853
                       or dropped.
 
854
      order_num        How many ORDER BY fields has been specified.
 
855
      order            List of fields to order_st BY.
 
856
      ignore           Whether we have ALTER IGNORE Table
 
857
 
 
858
  DESCRIPTION
 
859
    This is a veery long function and is everything but the kitchen sink :)
 
860
    It is used to alter a table and not only by ALTER Table but also
 
861
    CREATE|DROP INDEX are mapped on this function.
 
862
 
 
863
    When the ALTER Table statement just does a RENAME or ENABLE|DISABLE KEYS,
 
864
    or both, then this function short cuts its operation by renaming
 
865
    the table and/or enabling/disabling the keys. In this case, the FRM is
 
866
    not changed, directly by alter_table. However, if there is a
 
867
    RENAME + change of a field, or an index, the short cut is not used.
 
868
    See how `create_list` is used to generate the new FRM regarding the
 
869
    structure of the fields. The same is done for the indices of the table.
 
870
 
 
871
    Important is the fact, that this function tries to do as little work as
 
872
    possible, by finding out whether a intermediate table is needed to copy
 
873
    data into and when finishing the altering to use it as the original table.
 
874
    For this reason the function compare_tables() is called, which decides
 
875
    based on all kind of data how similar are the new and the original
 
876
    tables.
 
877
 
 
878
  RETURN VALUES
 
879
    false  OK
 
880
    true   Error
 
881
*/
 
882
 
 
883
static bool internal_alter_table(Session *session,
 
884
                                 Table *table,
 
885
                                 TableIdentifier &original_table_identifier,
 
886
                                 TableIdentifier &new_table_identifier,
 
887
                                 HA_CREATE_INFO *create_info,
 
888
                                 const message::Table &original_proto,
 
889
                                 message::Table &create_proto,
 
890
                                 TableList *table_list,
 
891
                                 AlterInfo *alter_info,
 
892
                                 uint32_t order_num,
 
893
                                 Order *order,
 
894
                                 bool ignore)
 
895
{
 
896
  int error= 0;
 
897
  char tmp_name[80];
 
898
  char old_name[32];
 
899
  ha_rows copied= 0;
 
900
  ha_rows deleted= 0;
 
901
 
 
902
  session->set_proc_info("init");
 
903
 
 
904
  table->use_all_columns();
 
905
 
 
906
  plugin::StorageEngine *new_engine;
 
907
  plugin::StorageEngine *original_engine;
 
908
 
 
909
  original_engine= table->getMutableShare()->getEngine();
 
910
 
 
911
  if (not create_info->db_type)
 
912
  {
 
913
    create_info->db_type= original_engine;
 
914
  }
 
915
  new_engine= create_info->db_type;
 
916
 
 
917
 
 
918
  create_proto.set_schema(new_table_identifier.getSchemaName());
 
919
  create_proto.set_type(new_table_identifier.getType());
 
920
 
 
921
  /**
 
922
    @todo Have a check on the table definition for FK in the future 
 
923
    to remove the need for the cursor. (aka can_switch_engines())
 
924
  */
 
925
  if (new_engine != original_engine &&
 
926
      not table->cursor->can_switch_engines())
 
927
  {
 
928
    assert(0);
 
929
    my_error(ER_ROW_IS_REFERENCED, MYF(0));
 
930
 
 
931
    return true;
 
932
  }
 
933
 
 
934
  if (original_engine->check_flag(HTON_BIT_ALTER_NOT_SUPPORTED) ||
 
935
      new_engine->check_flag(HTON_BIT_ALTER_NOT_SUPPORTED))
 
936
  {
 
937
    std::string path;
 
938
    new_table_identifier.getSQLPath(path);
 
939
    my_error(ER_ILLEGAL_HA, MYF(0), path.c_str());
 
940
 
 
941
    return true;
 
942
  }
 
943
 
 
944
  session->set_proc_info("setup");
 
945
 
 
946
  /*
 
947
   * test if no other bits except ALTER_RENAME and ALTER_KEYS_ONOFF are set
 
948
 */
 
949
  {
 
950
    bitset<32> tmp;
 
951
 
 
952
    tmp.set();
 
953
    tmp.reset(ALTER_RENAME);
 
954
    tmp.reset(ALTER_KEYS_ONOFF);
 
955
    tmp&= alter_info->flags;
 
956
 
 
957
    if (! (tmp.any()) && ! table->getShare()->getType()) // no need to touch frm
 
958
    {
 
959
      switch (alter_info->keys_onoff)
 
960
      {
 
961
      case LEAVE_AS_IS:
 
962
        break;
 
963
      case ENABLE:
 
964
        /*
 
965
          wait_while_table_is_used() ensures that table being altered is
 
966
          opened only by this thread and that Table::TableShare::version
 
967
          of Table object corresponding to this table is 0.
 
968
          The latter guarantees that no DML statement will open this table
 
969
          until ALTER Table finishes (i.e. until close_thread_tables())
 
970
          while the fact that the table is still open gives us protection
 
971
          from concurrent DDL statements.
 
972
        */
 
973
        table::Cache::singleton().mutex().lock(); /* DDL wait for/blocker */
 
974
        wait_while_table_is_used(session, table, HA_EXTRA_FORCE_REOPEN);
 
975
        table::Cache::singleton().mutex().unlock();
 
976
        error= table->cursor->ha_enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
 
977
        /* COND_refresh will be signaled in close_thread_tables() */
 
978
        break;
 
979
      case DISABLE:
 
980
        table::Cache::singleton().mutex().lock(); /* DDL wait for/blocker */
 
981
        wait_while_table_is_used(session, table, HA_EXTRA_FORCE_REOPEN);
 
982
        table::Cache::singleton().mutex().unlock();
 
983
        error=table->cursor->ha_disable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
 
984
        /* COND_refresh will be signaled in close_thread_tables() */
 
985
        break;
 
986
      default:
 
987
        assert(false);
 
988
        error= 0;
 
989
        break;
 
990
      }
 
991
 
 
992
      if (error == HA_ERR_WRONG_COMMAND)
 
993
      {
 
994
        error= 0;
 
995
        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
996
                            ER_ILLEGAL_HA, ER(ER_ILLEGAL_HA),
 
997
                            table->getAlias());
 
998
      }
 
999
 
 
1000
      table::Cache::singleton().mutex().lock(); /* Lock to remove all instances of table from table cache before ALTER */
 
1001
      /*
 
1002
        Unlike to the above case close_cached_table() below will remove ALL
 
1003
        instances of Table from table cache (it will also remove table lock
 
1004
        held by this thread). So to make actual table renaming and writing
 
1005
        to binlog atomic we have to put them into the same critical section
 
1006
        protected by table::Cache::singleton().mutex() mutex. This also removes gap for races between
 
1007
        access() and mysql_rename_table() calls.
 
1008
      */
 
1009
 
 
1010
      if (error == 0 &&  not (original_table_identifier == new_table_identifier))
 
1011
      {
 
1012
        session->set_proc_info("rename");
 
1013
        /*
 
1014
          Then do a 'simple' rename of the table. First we need to close all
 
1015
          instances of 'source' table.
 
1016
        */
 
1017
        session->close_cached_table(table);
 
1018
        /*
 
1019
          Then, we want check once again that target table does not exist.
 
1020
          Actually the order of these two steps does not matter since
 
1021
          earlier we took name-lock on the target table, so we do them
 
1022
          in this particular order only to be consistent with 5.0, in which
 
1023
          we don't take this name-lock and where this order really matters.
 
1024
          @todo Investigate if we need this access() check at all.
 
1025
        */
 
1026
        if (plugin::StorageEngine::doesTableExist(*session, new_table_identifier))
 
1027
        {
 
1028
          std::string path;
 
1029
          new_table_identifier.getSQLPath(path);
 
1030
          my_error(ER_TABLE_EXISTS_ERROR, MYF(0), path.c_str());
 
1031
          error= -1;
 
1032
        }
 
1033
        else
 
1034
        {
 
1035
          if (mysql_rename_table(*session, original_engine, original_table_identifier, new_table_identifier))
 
1036
          {
 
1037
            error= -1;
 
1038
          }
 
1039
        }
 
1040
      }
 
1041
 
 
1042
      if (error == HA_ERR_WRONG_COMMAND)
 
1043
      {
 
1044
        error= 0;
 
1045
        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
1046
                            ER_ILLEGAL_HA, ER(ER_ILLEGAL_HA),
 
1047
                            table->getAlias());
 
1048
      }
 
1049
 
 
1050
      if (error == 0)
 
1051
      {
 
1052
        TransactionServices &transaction_services= TransactionServices::singleton();
 
1053
        transaction_services.allocateNewTransactionId();
 
1054
        write_bin_log(session, *session->getQueryString());
 
1055
        session->my_ok();
 
1056
      }
 
1057
      else if (error > 0)
 
1058
      {
 
1059
        table->print_error(error, MYF(0));
 
1060
        error= -1;
 
1061
      }
 
1062
 
 
1063
      table::Cache::singleton().mutex().unlock();
 
1064
      table_list->table= NULL;
 
1065
 
 
1066
      return error;
 
1067
    }
 
1068
  }
 
1069
 
 
1070
  /* We have to do full alter table. */
 
1071
  new_engine= create_info->db_type;
 
1072
 
 
1073
  if (mysql_prepare_alter_table(session, table, create_info, original_proto, create_proto, alter_info))
 
1074
  {
 
1075
    return true;
 
1076
  }
 
1077
 
 
1078
  set_table_default_charset(create_info, new_table_identifier.getSchemaName().c_str());
 
1079
 
 
1080
  alter_info->build_method= HA_BUILD_OFFLINE;
 
1081
 
 
1082
  snprintf(tmp_name, sizeof(tmp_name), "%s-%lx_%"PRIx64, TMP_FILE_PREFIX, (unsigned long) current_pid, session->thread_id);
 
1083
 
 
1084
  /* Create a temporary table with the new format */
 
1085
  /**
 
1086
    @note we make an internal temporary table unless the table is a temporary table. In last
 
1087
    case we just use it as is. Neither of these tables require locks in order to  be
 
1088
    filled.
 
1089
  */
 
1090
  TableIdentifier new_table_as_temporary(original_table_identifier.getSchemaName(),
 
1091
                                         tmp_name,
 
1092
                                         create_proto.type() != message::Table::TEMPORARY ? message::Table::INTERNAL :
 
1093
                                         message::Table::TEMPORARY);
 
1094
 
 
1095
  error= create_temporary_table(session, new_table_as_temporary, create_info, create_proto, alter_info);
 
1096
 
 
1097
  if (error != 0)
 
1098
  {
 
1099
    return true;
 
1100
  }
 
1101
 
 
1102
  /* Open the table so we need to copy the data to it. */
 
1103
  Table *new_table= open_alter_table(session, table, new_table_as_temporary);
 
1104
 
 
1105
 
 
1106
  if (not new_table)
 
1107
  {
 
1108
    plugin::StorageEngine::dropTable(*session, new_table_as_temporary);
 
1109
    return true;
 
1110
  }
 
1111
 
 
1112
  /* Copy the data if necessary. */
 
1113
  {
 
1114
    /* We must not ignore bad input! */
 
1115
    session->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;    // calc cuted fields
 
1116
    session->cuted_fields= 0L;
 
1117
    session->set_proc_info("copy to tmp table");
 
1118
    copied= deleted= 0;
 
1119
 
 
1120
    /* We don't want update TIMESTAMP fields during ALTER Table. */
 
1121
    new_table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
 
1122
    new_table->next_number_field= new_table->found_next_number_field;
 
1123
    error= copy_data_between_tables(session,
 
1124
                                    table,
 
1125
                                    new_table,
 
1126
                                    alter_info->create_list,
 
1127
                                    ignore,
 
1128
                                    order_num,
 
1129
                                    order,
 
1130
                                    &copied,
 
1131
                                    &deleted,
 
1132
                                    alter_info->keys_onoff,
 
1133
                                    alter_info->error_if_not_empty);
 
1134
 
 
1135
    /* We must not ignore bad input! */
 
1136
    assert(session->count_cuted_fields == CHECK_FIELD_ERROR_FOR_NULL);
 
1137
  }
 
1138
 
 
1139
  /* Now we need to resolve what just happened with the data copy. */
 
1140
 
 
1141
  if (error)
 
1142
  {
 
1143
 
 
1144
    /*
 
1145
      No default value was provided for a DATE/DATETIME field, the
 
1146
      current sql_mode doesn't allow the '0000-00-00' value and
 
1147
      the table to be altered isn't empty.
 
1148
      Report error here.
 
1149
    */
 
1150
    if (alter_info->error_if_not_empty && session->row_count)
 
1151
    {
 
1152
      my_error(ER_INVALID_ALTER_TABLE_FOR_NOT_NULL, MYF(0));
 
1153
    }
 
1154
 
 
1155
    if (original_table_identifier.isTmp())
 
1156
    {
 
1157
      if (new_table)
 
1158
      {
 
1159
        /* close_temporary_table() frees the new_table pointer. */
 
1160
        session->close_temporary_table(new_table);
 
1161
      }
 
1162
      else
 
1163
      {
 
1164
        plugin::StorageEngine::dropTable(*session, new_table_as_temporary);
 
1165
      }
 
1166
 
 
1167
      return true;
 
1168
    }
 
1169
    else
 
1170
    {
 
1171
      if (new_table)
 
1172
      {
 
1173
        /*
 
1174
          Close the intermediate table that will be the new table.
 
1175
          Note that MERGE tables do not have their children attached here.
 
1176
        */
 
1177
        new_table->intern_close_table();
 
1178
        if (new_table->hasShare())
 
1179
        {
 
1180
          delete new_table->getMutableShare();
 
1181
        }
 
1182
 
 
1183
        delete new_table;
 
1184
      }
 
1185
 
 
1186
      table::Cache::singleton().mutex().lock(); /* ALTER TABLE */
 
1187
 
 
1188
      plugin::StorageEngine::dropTable(*session, new_table_as_temporary);
 
1189
      table::Cache::singleton().mutex().unlock();
 
1190
 
 
1191
      return true;
 
1192
    }
 
1193
  }
 
1194
  // Temporary table and success
 
1195
  else if (original_table_identifier.isTmp())
 
1196
  {
 
1197
    /* Close lock if this is a transactional table */
 
1198
    if (session->lock)
 
1199
    {
 
1200
      session->unlockTables(session->lock);
 
1201
      session->lock= 0;
 
1202
    }
 
1203
 
 
1204
    /* Remove link to old table and rename the new one */
 
1205
    session->close_temporary_table(table);
 
1206
 
 
1207
    /* Should pass the 'new_name' as we store table name in the cache */
 
1208
    new_table->getMutableShare()->setIdentifier(new_table_identifier);
 
1209
 
 
1210
    new_table_identifier.setPath(new_table_as_temporary.getPath());
 
1211
 
 
1212
    if (mysql_rename_table(*session, new_engine, new_table_as_temporary, new_table_identifier) != 0)
 
1213
    {
 
1214
      return true;
 
1215
    }
 
1216
  }
 
1217
  // Normal table success
 
1218
  else
 
1219
  {
 
1220
    if (new_table)
 
1221
    {
 
1222
      /*
 
1223
        Close the intermediate table that will be the new table.
 
1224
        Note that MERGE tables do not have their children attached here.
 
1225
      */
 
1226
      new_table->intern_close_table();
 
1227
 
 
1228
      if (new_table->hasShare())
 
1229
      {
 
1230
        delete new_table->getMutableShare();
 
1231
      }
 
1232
 
 
1233
      delete new_table;
 
1234
    }
 
1235
 
 
1236
    table::Cache::singleton().mutex().lock(); /* ALTER TABLE */
 
1237
 
 
1238
    /*
 
1239
      Data is copied. Now we:
 
1240
      1) Wait until all other threads close old version of table.
 
1241
      2) Close instances of table open by this thread and replace them
 
1242
      with exclusive name-locks.
 
1243
      3) Rename the old table to a temp name, rename the new one to the
 
1244
      old name.
 
1245
      4) If we are under LOCK TABLES and don't do ALTER Table ... RENAME
 
1246
      we reopen new version of table.
 
1247
      5) Write statement to the binary log.
 
1248
      6) If we are under LOCK TABLES and do ALTER Table ... RENAME we
 
1249
      remove name-locks from list of open tables and table cache.
 
1250
      7) If we are not not under LOCK TABLES we rely on close_thread_tables()
 
1251
      call to remove name-locks from table cache and list of open table.
 
1252
    */
 
1253
 
 
1254
    session->set_proc_info("rename result table");
 
1255
 
 
1256
    snprintf(old_name, sizeof(old_name), "%s2-%lx-%"PRIx64, TMP_FILE_PREFIX, (unsigned long) current_pid, session->thread_id);
 
1257
 
 
1258
    my_casedn_str(files_charset_info, old_name);
 
1259
 
 
1260
    wait_while_table_is_used(session, table, HA_EXTRA_PREPARE_FOR_RENAME);
 
1261
    session->close_data_files_and_morph_locks(original_table_identifier);
 
1262
 
 
1263
    error= 0;
 
1264
 
 
1265
    /*
 
1266
      This leads to the storage engine (SE) not being notified for renames in
 
1267
      mysql_rename_table(), because we just juggle with the FRM and nothing
 
1268
      more. If we have an intermediate table, then we notify the SE that
 
1269
      it should become the actual table. Later, we will recycle the old table.
 
1270
      However, in case of ALTER Table RENAME there might be no intermediate
 
1271
      table. This is when the old and new tables are compatible, according to
 
1272
      compare_table(). Then, we need one additional call to
 
1273
    */
 
1274
    TableIdentifier original_table_to_drop(original_table_identifier.getSchemaName(),
 
1275
                                           old_name, create_proto.type() != message::Table::TEMPORARY ? message::Table::INTERNAL :
 
1276
                                         message::Table::TEMPORARY);
 
1277
 
 
1278
    if (mysql_rename_table(*session, original_engine, original_table_identifier, original_table_to_drop))
 
1279
    {
 
1280
      error= 1;
 
1281
      plugin::StorageEngine::dropTable(*session, new_table_as_temporary);
 
1282
    }
 
1283
    else
 
1284
    {
 
1285
      if (mysql_rename_table(*session, new_engine, new_table_as_temporary, new_table_identifier) != 0)
 
1286
      {
 
1287
        /* Try to get everything back. */
 
1288
        error= 1;
 
1289
 
 
1290
        plugin::StorageEngine::dropTable(*session, new_table_identifier);
 
1291
 
 
1292
        plugin::StorageEngine::dropTable(*session, new_table_as_temporary);
 
1293
 
 
1294
        mysql_rename_table(*session, original_engine, original_table_to_drop, original_table_identifier);
 
1295
      }
 
1296
      else
 
1297
      {
 
1298
        plugin::StorageEngine::dropTable(*session, original_table_to_drop);
 
1299
      }
 
1300
    }
 
1301
 
 
1302
    if (error)
 
1303
    {
 
1304
      /*
 
1305
        An error happened while we were holding exclusive name-lock on table
 
1306
        being altered. To be safe under LOCK TABLES we should remove placeholders
 
1307
        from list of open tables list and table cache.
 
1308
      */
 
1309
      session->unlink_open_table(table);
 
1310
      table::Cache::singleton().mutex().unlock();
 
1311
 
 
1312
      return true;
 
1313
    }
 
1314
 
 
1315
    table::Cache::singleton().mutex().unlock();
 
1316
 
 
1317
    session->set_proc_info("end");
 
1318
 
 
1319
    write_bin_log(session, *session->getQueryString());
 
1320
    table_list->table= NULL;
 
1321
  }
 
1322
 
 
1323
  /*
 
1324
   * Field::store() may have called my_error().  If this is 
 
1325
   * the case, we must not send an ok packet, since 
 
1326
   * Diagnostics_area::is_set() will fail an assert.
 
1327
 */
 
1328
  if (session->is_error())
 
1329
  {
 
1330
    /* my_error() was called.  Return true (which means error...) */
 
1331
    return true;
 
1332
  }
 
1333
 
 
1334
  snprintf(tmp_name, sizeof(tmp_name), ER(ER_INSERT_INFO),
 
1335
           (ulong) (copied + deleted), (ulong) deleted,
 
1336
           (ulong) session->cuted_fields);
 
1337
  session->my_ok(copied + deleted, 0, 0L, tmp_name);
 
1338
  session->some_tables_deleted= 0;
 
1339
 
 
1340
  return false;
 
1341
}
 
1342
 
 
1343
bool alter_table(Session *session,
 
1344
                 TableIdentifier &original_table_identifier,
 
1345
                 TableIdentifier &new_table_identifier,
 
1346
                 HA_CREATE_INFO *create_info,
 
1347
                 const message::Table &original_proto,
 
1348
                 message::Table &create_proto,
 
1349
                 TableList *table_list,
 
1350
                 AlterInfo *alter_info,
 
1351
                 uint32_t order_num,
 
1352
                 Order *order,
 
1353
                 bool ignore)
 
1354
{
 
1355
  bool error;
 
1356
  Table *table;
 
1357
 
 
1358
  if (alter_info->tablespace_op != NO_TABLESPACE_OP)
 
1359
  {
 
1360
    /* DISCARD/IMPORT TABLESPACE is always alone in an ALTER Table */
 
1361
    return mysql_discard_or_import_tablespace(session, table_list, alter_info->tablespace_op);
 
1362
  }
 
1363
 
 
1364
  session->set_proc_info("init");
 
1365
 
 
1366
  if (not (table= session->openTableLock(table_list, TL_WRITE_ALLOW_READ)))
 
1367
    return true;
 
1368
 
 
1369
  session->set_proc_info("gained write lock on table");
 
1370
 
 
1371
  /* 
 
1372
    Check that we are not trying to rename to an existing table,
 
1373
    if one existed we get a lock, if we can't we error.
 
1374
  */
 
1375
  {
 
1376
    Table *name_lock= NULL;
 
1377
 
 
1378
    if (not lockTableIfDifferent(*session, original_table_identifier, new_table_identifier, name_lock))
 
1379
    {
 
1380
      return true;
 
1381
    }
 
1382
 
 
1383
    error= internal_alter_table(session,
 
1384
                                table,
 
1385
                                original_table_identifier,
 
1386
                                new_table_identifier,
 
1387
                                create_info,
 
1388
                                original_proto,
 
1389
                                create_proto,
 
1390
                                table_list,
 
1391
                                alter_info,
 
1392
                                order_num,
 
1393
                                order,
 
1394
                                ignore);
 
1395
 
 
1396
    if (name_lock)
 
1397
    {
 
1398
      table::Cache::singleton().mutex().lock(); /* ALTER TABLe */
 
1399
      session->unlink_open_table(name_lock);
 
1400
      table::Cache::singleton().mutex().unlock();
 
1401
    }
 
1402
  }
 
1403
 
 
1404
  return error;
 
1405
}
 
1406
/* alter_table */
 
1407
 
 
1408
static int
 
1409
copy_data_between_tables(Session *session,
 
1410
                         Table *from, Table *to,
 
1411
                         List<CreateField> &create,
 
1412
                         bool ignore,
 
1413
                         uint32_t order_num, Order *order,
 
1414
                         ha_rows *copied,
 
1415
                         ha_rows *deleted,
 
1416
                         enum enum_enable_or_disable keys_onoff,
 
1417
                         bool error_if_not_empty)
 
1418
{
 
1419
  int error= 0;
 
1420
  CopyField *copy,*copy_end;
 
1421
  ulong found_count,delete_count;
 
1422
  uint32_t length= 0;
 
1423
  SortField *sortorder;
 
1424
  ReadRecord info;
 
1425
  TableList   tables;
 
1426
  List<Item>   fields;
 
1427
  List<Item>   all_fields;
 
1428
  ha_rows examined_rows;
 
1429
  bool auto_increment_field_copied= 0;
 
1430
  uint64_t prev_insert_id;
 
1431
 
 
1432
  /*
 
1433
    Turn off recovery logging since rollback of an alter table is to
 
1434
    delete the new table so there is no need to log the changes to it.
 
1435
 
 
1436
    This needs to be done before external_lock
 
1437
  */
 
1438
  TransactionServices &transaction_services= TransactionServices::singleton();
 
1439
 
 
1440
  /* 
 
1441
   * LP Bug #552420 
 
1442
   *
 
1443
   * Since open_temporary_table() doesn't invoke lockTables(), we
 
1444
   * don't get the usual automatic call to StorageEngine::startStatement(), so
 
1445
   * we manually call it here...
 
1446
   */
 
1447
  to->getMutableShare()->getEngine()->startStatement(session);
 
1448
 
 
1449
  if (!(copy= new CopyField[to->getShare()->sizeFields()]))
 
1450
    return -1;
 
1451
 
 
1452
  if (to->cursor->ha_external_lock(session, F_WRLCK))
 
1453
    return -1;
 
1454
 
 
1455
  /* We need external lock before we can disable/enable keys */
 
1456
  alter_table_manage_keys(session, to, from->cursor->indexes_are_disabled(), keys_onoff);
 
1457
 
 
1458
  /* We can abort alter table for any table type */
 
1459
  session->abort_on_warning= !ignore;
 
1460
 
 
1461
  from->cursor->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
 
1462
  to->cursor->ha_start_bulk_insert(from->cursor->stats.records);
 
1463
 
 
1464
  List_iterator<CreateField> it(create);
 
1465
  CreateField *def;
 
1466
  copy_end=copy;
 
1467
  for (Field **ptr= to->getFields(); *ptr ; ptr++)
 
1468
  {
 
1469
    def=it++;
 
1470
    if (def->field)
 
1471
    {
 
1472
      if (*ptr == to->next_number_field)
 
1473
        auto_increment_field_copied= true;
 
1474
 
 
1475
      (copy_end++)->set(*ptr,def->field,0);
 
1476
    }
 
1477
 
 
1478
  }
 
1479
 
 
1480
  found_count=delete_count=0;
 
1481
 
 
1482
  if (order)
 
1483
  {
 
1484
    if (to->getShare()->hasPrimaryKey() && to->cursor->primary_key_is_clustered())
 
1485
    {
 
1486
      char warn_buff[DRIZZLE_ERRMSG_SIZE];
 
1487
      snprintf(warn_buff, sizeof(warn_buff),
 
1488
               _("order_st BY ignored because there is a user-defined clustered "
 
1489
                 "index in the table '%-.192s'"),
 
1490
               from->getMutableShare()->getTableName());
 
1491
      push_warning(session, DRIZZLE_ERROR::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR,
 
1492
                   warn_buff);
 
1493
    }
 
1494
    else
 
1495
    {
 
1496
      FileSort filesort(*session);
 
1497
      from->sort.io_cache= new internal::IO_CACHE;
 
1498
 
 
1499
      memset(&tables, 0, sizeof(tables));
 
1500
      tables.table= from;
 
1501
      tables.setTableName(const_cast<char *>(from->getMutableShare()->getTableName()));
 
1502
      tables.alias= const_cast<char *>(tables.getTableName());
 
1503
      tables.setSchemaName(const_cast<char *>(from->getMutableShare()->getSchemaName()));
 
1504
      error= 1;
 
1505
 
 
1506
      if (session->lex->select_lex.setup_ref_array(session, order_num) ||
 
1507
          setup_order(session, session->lex->select_lex.ref_pointer_array,
 
1508
                      &tables, fields, all_fields, order) ||
 
1509
          !(sortorder= make_unireg_sortorder(order, &length, NULL)) ||
 
1510
          (from->sort.found_records= filesort.run(from, sortorder, length,
 
1511
                                                  (optimizer::SqlSelect *) 0, HA_POS_ERROR,
 
1512
                                                  1, examined_rows)) == HA_POS_ERROR)
 
1513
      {
 
1514
        goto err;
 
1515
      }
 
1516
    }
 
1517
  }
 
1518
 
 
1519
  /* Tell handler that we have values for all columns in the to table */
 
1520
  to->use_all_columns();
 
1521
  info.init_read_record(session, from, (optimizer::SqlSelect *) 0, 1, true);
 
1522
  if (ignore)
 
1523
    to->cursor->extra(HA_EXTRA_IGNORE_DUP_KEY);
 
1524
  session->row_count= 0;
 
1525
  to->restoreRecordAsDefault();        // Create empty record
 
1526
  while (!(error=info.read_record(&info)))
 
1527
  {
 
1528
    if (session->getKilled())
 
1529
    {
 
1530
      session->send_kill_message();
 
1531
      error= 1;
 
1532
      break;
 
1533
    }
 
1534
    session->row_count++;
 
1535
    /* Return error if source table isn't empty. */
 
1536
    if (error_if_not_empty)
 
1537
    {
 
1538
      error= 1;
 
1539
      break;
 
1540
    }
 
1541
    if (to->next_number_field)
 
1542
    {
 
1543
      if (auto_increment_field_copied)
 
1544
        to->auto_increment_field_not_null= true;
 
1545
      else
 
1546
        to->next_number_field->reset();
 
1547
    }
 
1548
 
 
1549
    for (CopyField *copy_ptr= copy; copy_ptr != copy_end ; copy_ptr++)
 
1550
    {
 
1551
      if (not copy->to_field->hasDefault() and copy->from_null_ptr and  *copy->from_null_ptr & copy->from_bit)
 
1552
      {
 
1553
        copy->to_field->set_warning(DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
1554
                                    ER_WARN_DATA_TRUNCATED, 1);
 
1555
        copy->to_field->reset();
 
1556
        error= 1;
 
1557
        break;
 
1558
      }
 
1559
 
 
1560
      copy_ptr->do_copy(copy_ptr);
 
1561
    }
 
1562
 
 
1563
    if (error)
 
1564
    {
 
1565
      break;
 
1566
    }
 
1567
 
 
1568
    prev_insert_id= to->cursor->next_insert_id;
 
1569
    error= to->cursor->insertRecord(to->record[0]);
 
1570
    to->auto_increment_field_not_null= false;
 
1571
 
 
1572
    if (error)
 
1573
    { 
 
1574
      if (!ignore || to->cursor->is_fatal_error(error, HA_CHECK_DUP))
 
1575
      { 
 
1576
        to->print_error(error, MYF(0));
 
1577
        break;
 
1578
      }
 
1579
      to->cursor->restore_auto_increment(prev_insert_id);
 
1580
      delete_count++;
 
1581
    }
 
1582
    else
 
1583
    {
 
1584
      found_count++;
 
1585
    }
 
1586
  }
 
1587
 
 
1588
  info.end_read_record();
 
1589
  from->free_io_cache();
 
1590
  delete [] copy;                               // This is never 0
 
1591
 
 
1592
  if (to->cursor->ha_end_bulk_insert() && error <= 0)
 
1593
  {
 
1594
    to->print_error(errno, MYF(0));
 
1595
    error=1;
 
1596
  }
 
1597
  to->cursor->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
1598
 
 
1599
  /*
 
1600
    Ensure that the new table is saved properly to disk so that we
 
1601
    can do a rename
 
1602
  */
 
1603
  if (transaction_services.autocommitOrRollback(session, false))
 
1604
    error=1;
 
1605
  if (! session->endActiveTransaction())
 
1606
    error=1;
 
1607
 
 
1608
 err:
 
1609
  session->abort_on_warning= 0;
 
1610
  from->free_io_cache();
 
1611
  *copied= found_count;
 
1612
  *deleted=delete_count;
 
1613
  to->cursor->ha_release_auto_increment();
 
1614
  if (to->cursor->ha_external_lock(session,F_UNLCK))
 
1615
    error=1;
 
1616
 
 
1617
  return(error > 0 ? -1 : 0);
 
1618
}
 
1619
 
 
1620
static int
 
1621
create_temporary_table(Session *session,
 
1622
                       TableIdentifier &identifier,
 
1623
                       HA_CREATE_INFO *create_info,
 
1624
                       message::Table &create_proto,
 
1625
                       AlterInfo *alter_info)
 
1626
{
 
1627
  int error;
 
1628
 
 
1629
  /*
 
1630
    Create a table with a temporary name.
 
1631
    We don't log the statement, it will be logged later.
 
1632
  */
 
1633
  create_proto.set_name(identifier.getTableName());
 
1634
 
 
1635
  create_proto.mutable_engine()->set_name(create_info->db_type->getName());
 
1636
 
 
1637
  error= mysql_create_table(session,
 
1638
                            identifier,
 
1639
                            create_info, create_proto, alter_info, true, 0, false);
 
1640
 
 
1641
  return error;
 
1642
}
 
1643
 
 
1644
static Table *open_alter_table(Session *session, Table *table, TableIdentifier &identifier)
 
1645
{
 
1646
  Table *new_table;
 
1647
 
 
1648
  /* Open the table so we need to copy the data to it. */
 
1649
  if (table->getShare()->getType())
 
1650
  {
 
1651
    TableList tbl;
 
1652
    tbl.setSchemaName(const_cast<char *>(identifier.getSchemaName().c_str()));
 
1653
    tbl.alias= const_cast<char *>(identifier.getTableName().c_str());
 
1654
    tbl.setTableName(const_cast<char *>(identifier.getTableName().c_str()));
 
1655
 
 
1656
    /* Table is in session->temporary_tables */
 
1657
    new_table= session->openTable(&tbl, (bool*) 0, DRIZZLE_LOCK_IGNORE_FLUSH);
 
1658
  }
 
1659
  else
 
1660
  {
 
1661
    /* Open our intermediate table */
 
1662
    new_table= session->open_temporary_table(identifier, false);
 
1663
  }
 
1664
 
 
1665
  return new_table;
 
1666
}
 
1667
 
 
1668
} /* namespace drizzled */