sql_insert.cc 119 KB
Newer Older
unknown's avatar
unknown committed
1
/* Copyright (C) 2000-2006 MySQL AB
unknown's avatar
unknown committed
2

unknown's avatar
unknown committed
3 4
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
unknown's avatar
unknown committed
5
   the Free Software Foundation; version 2 of the License.
unknown's avatar
unknown committed
6

unknown's avatar
unknown committed
7 8 9 10
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
unknown's avatar
unknown committed
11

unknown's avatar
unknown committed
12 13 14 15 16 17 18
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


/* Insert of records */

19 20 21 22 23 24 25 26 27 28 29 30 31
/*
  INSERT DELAYED

  Insert delayed is distinguished from a normal insert by lock_type ==
  TL_WRITE_DELAYED instead of TL_WRITE. It first tries to open a
  "delayed" table (delayed_get_table()), but falls back to
  open_and_lock_tables() on error and proceeds as normal insert then.

  Opening a "delayed" table means to find a delayed insert thread that
  has the table open already. If this fails, a new thread is created and
  waited for to open and lock the table.

  If accessing the thread succeeded, in
32
  Delayed_insert::get_local_table() the table of the thread is copied
33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56
  for local use. A copy is required because the normal insert logic
  works on a target table, but the other threads table object must not
  be used. The insert logic uses the record buffer to create a record.
  And the delayed insert thread uses the record buffer to pass the
  record to the table handler. So there must be different objects. Also
  the copied table is not included in the lock, so that the statement
  can proceed even if the real table cannot be accessed at this moment.

  Copying a table object is not a trivial operation. Besides the TABLE
  object there are the field pointer array, the field objects and the
  record buffer. After copying the field objects, their pointers into
  the record must be "moved" to point to the new record buffer.

  After this setup the normal insert logic is used. Only that for
  delayed inserts write_delayed() is called instead of write_record().
  It inserts the rows into a queue and signals the delayed insert thread
  instead of writing directly to the table.

  The delayed insert thread awakes from the signal. It locks the table,
  inserts the rows from the queue, unlocks the table, and waits for the
  next signal. It does normally live until a FLUSH TABLES or SHUTDOWN.

*/

unknown's avatar
unknown committed
57
#include "mysql_priv.h"
58 59
#include "sp_head.h"
#include "sql_trigger.h"
60
#include "sql_select.h"
61
#include "sql_show.h"
62
#include "slave.h"
63
#include "rpl_mi.h"
unknown's avatar
unknown committed
64

unknown's avatar
unknown committed
65
#ifndef EMBEDDED_LIBRARY
unknown's avatar
unknown committed
66
static bool delayed_get_table(THD *thd, TABLE_LIST *table_list);
unknown's avatar
unknown committed
67
static int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
68
                         LEX_STRING query, bool ignore, bool log_on);
unknown's avatar
unknown committed
69
static void end_delayed_insert(THD *thd);
70
pthread_handler_t handle_delayed_insert(void *arg);
unknown's avatar
unknown committed
71
static void unlink_blobs(register TABLE *table);
unknown's avatar
unknown committed
72
#endif
73
static bool check_view_insertability(THD *thd, TABLE_LIST *view);
unknown's avatar
unknown committed
74 75 76 77 78 79 80 81 82 83 84

/* Define to force use of my_malloc() if the allocated memory block is big */

#ifndef HAVE_ALLOCA
#define my_safe_alloca(size, min_length) my_alloca(size)
#define my_safe_afree(ptr, size, min_length) my_afree(ptr)
#else
#define my_safe_alloca(size, min_length) ((size <= min_length) ? my_alloca(size) : my_malloc(size,MYF(0)))
#define my_safe_afree(ptr, size, min_length) if (size > min_length) my_free(ptr,MYF(0))
#endif

85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143
/*
  Check that insert/update fields are from the same single table of a view.

  SYNOPSIS
    check_view_single_update()
    fields            The insert/update fields to be checked.
    view              The view for insert.
    map     [in/out]  The insert table map.

  DESCRIPTION
    This function is called in 2 cases:
    1. to check insert fields. In this case *map will be set to 0.
       Insert fields are checked to be all from the same single underlying
       table of the given view. Otherwise the error is thrown. Found table
       map is returned in the map parameter.
    2. to check update fields of the ON DUPLICATE KEY UPDATE clause.
       In this case *map contains table_map found on the previous call of
       the function to check insert fields. Update fields are checked to be
       from the same table as the insert fields.

  RETURN
    0   OK
    1   Error
*/

bool check_view_single_update(List<Item> &fields, TABLE_LIST *view,
                              table_map *map)
{
  /* it is join view => we need to find the table for update */
  List_iterator_fast<Item> it(fields);
  Item *item;
  TABLE_LIST *tbl= 0;            // reset for call to check_single_table()
  table_map tables= 0;

  while ((item= it++))
    tables|= item->used_tables();

  /* Check found map against provided map */
  if (*map)
  {
    if (tables != *map)
      goto error;
    return FALSE;
  }

  if (view->check_single_table(&tbl, tables, view) || tbl == 0)
    goto error;

  view->table= tbl->table;
  *map= tables;

  return FALSE;

error:
  my_error(ER_VIEW_MULTIUPDATE, MYF(0),
           view->view_db.str, view->view_name.str);
  return TRUE;
}

144

unknown's avatar
unknown committed
145
/*
146
  Check if insert fields are correct.
147 148 149 150 151 152 153

  SYNOPSIS
    check_insert_fields()
    thd                         The current thread.
    table                       The table for insert.
    fields                      The insert fields.
    values                      The insert values.
unknown's avatar
unknown committed
154
    check_unique                If duplicate values should be rejected.
155 156 157 158 159 160 161 162 163

  NOTE
    Clears TIMESTAMP_AUTO_SET_ON_INSERT from table->timestamp_field_type
    or leaves it as is, depending on if timestamp should be updated or
    not.

  RETURN
    0           OK
    -1          Error
unknown's avatar
unknown committed
164 165
*/

unknown's avatar
unknown committed
166 167
static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
                               List<Item> &fields, List<Item> &values,
168
                               bool check_unique, table_map *map)
unknown's avatar
unknown committed
169
{
unknown's avatar
VIEW  
unknown committed
170
  TABLE *table= table_list->table;
171

172 173
  if (!table_list->updatable)
  {
174
    my_error(ER_NON_INSERTABLE_TABLE, MYF(0), table_list->alias, "INSERT");
175 176 177
    return -1;
  }

unknown's avatar
unknown committed
178 179
  if (fields.elements == 0 && values.elements != 0)
  {
180 181 182 183 184 185
    if (!table)
    {
      my_error(ER_VIEW_NO_INSERT_FIELD_LIST, MYF(0),
               table_list->view_db.str, table_list->view_name.str);
      return -1;
    }
186
    if (values.elements != table->s->fields)
unknown's avatar
unknown committed
187
    {
unknown's avatar
unknown committed
188
      my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1L);
unknown's avatar
unknown committed
189 190
      return -1;
    }
unknown's avatar
unknown committed
191
#ifndef NO_EMBEDDED_ACCESS_CHECKS
192 193 194 195 196 197
    Field_iterator_table field_it;
    field_it.set_table(table);
    if (check_grant_all_columns(thd, INSERT_ACL, &table->grant,
                                table->s->db.str, table->s->table_name.str,
                                &field_it))
      return -1;
unknown's avatar
unknown committed
198
#endif
199 200
    clear_timestamp_auto_bits(table->timestamp_field_type,
                              TIMESTAMP_AUTO_SET_ON_INSERT);
201 202 203 204
    /*
      No fields are provided so all fields must be provided in the values.
      Thus we set all bits in the write set.
    */
205
    bitmap_set_all(table->write_set);
unknown's avatar
unknown committed
206 207 208
  }
  else
  {						// Part field list
unknown's avatar
unknown committed
209 210
    SELECT_LEX *select_lex= &thd->lex->select_lex;
    Name_resolution_context *context= &select_lex->context;
211
    Name_resolution_context_state ctx_state;
unknown's avatar
unknown committed
212
    int res;
unknown's avatar
unknown committed
213

unknown's avatar
unknown committed
214 215
    if (fields.elements != values.elements)
    {
unknown's avatar
unknown committed
216
      my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1L);
unknown's avatar
unknown committed
217 218 219
      return -1;
    }

220
    thd->dup_field= 0;
unknown's avatar
unknown committed
221 222 223
    select_lex->no_wrap_view_item= TRUE;

    /* Save the state of the current name resolution context. */
224
    ctx_state.save_state(context, table_list);
unknown's avatar
unknown committed
225 226 227 228 229

    /*
      Perform name resolution only in the first table - 'table_list',
      which is the table that is inserted into.
    */
230
    table_list->next_local= 0;
231
    context->resolve_in_table_list_only(table_list);
232
    res= setup_fields(thd, 0, fields, MARK_COLUMNS_WRITE, 0, 0);
unknown's avatar
unknown committed
233 234

    /* Restore the current context. */
235
    ctx_state.restore_state(context, table_list);
236
    thd->lex->select_lex.no_wrap_view_item= FALSE;
unknown's avatar
unknown committed
237

unknown's avatar
unknown committed
238
    if (res)
unknown's avatar
unknown committed
239
      return -1;
unknown's avatar
unknown committed
240

unknown's avatar
unknown committed
241
    if (table_list->effective_algorithm == VIEW_ALGORITHM_MERGE)
242
    {
243
      if (check_view_single_update(fields, table_list, map))
244
        return -1;
245
      table= table_list->table;
246
    }
247

248
    if (check_unique && thd->dup_field)
unknown's avatar
unknown committed
249
    {
250
      my_error(ER_FIELD_SPECIFIED_TWICE, MYF(0), thd->dup_field->field_name);
unknown's avatar
unknown committed
251 252
      return -1;
    }
253 254 255 256 257 258 259 260 261 262 263 264
    if (table->timestamp_field)	// Don't automaticly set timestamp if used
    {
      if (bitmap_is_set(table->write_set,
                        table->timestamp_field->field_index))
        clear_timestamp_auto_bits(table->timestamp_field_type,
                                  TIMESTAMP_AUTO_SET_ON_INSERT);
      else
      {
        bitmap_set_bit(table->write_set,
                       table->timestamp_field->field_index);
      }
    }
unknown's avatar
unknown committed
265
  }
unknown's avatar
unknown committed
266
  // For the values we need select_priv
unknown's avatar
unknown committed
267
#ifndef NO_EMBEDDED_ACCESS_CHECKS
268
  table->grant.want_privilege= (SELECT_ACL & ~table->grant.privilege);
unknown's avatar
unknown committed
269
#endif
270 271 272

  if (check_key_in_view(thd, table_list) ||
      (table_list->view &&
273
       check_view_insertability(thd, table_list)))
274
  {
275
    my_error(ER_NON_INSERTABLE_TABLE, MYF(0), table_list->alias, "INSERT");
276 277 278
    return -1;
  }

unknown's avatar
unknown committed
279 280 281 282
  return 0;
}


283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301
/*
  Check update fields for the timestamp field.

  SYNOPSIS
    check_update_fields()
    thd                         The current thread.
    insert_table_list           The insert table list.
    table                       The table for update.
    update_fields               The update fields.

  NOTE
    If the update fields include the timestamp field,
    remove TIMESTAMP_AUTO_SET_ON_UPDATE from table->timestamp_field_type.

  RETURN
    0           OK
    -1          Error
*/

unknown's avatar
unknown committed
302
static int check_update_fields(THD *thd, TABLE_LIST *insert_table_list,
303
                               List<Item> &update_fields, table_map *map)
304
{
unknown's avatar
unknown committed
305
  TABLE *table= insert_table_list->table;
306
  my_bool timestamp_mark;
307

308 309
  LINT_INIT(timestamp_mark);

310 311
  if (table->timestamp_field)
  {
312 313 314 315 316 317
    /*
      Unmark the timestamp field so that we can check if this is modified
      by update_fields
    */
    timestamp_mark= bitmap_test_and_clear(table->write_set,
                                          table->timestamp_field->field_index);
318 319
  }

320 321
  /* Check the fields we are going to modify */
  if (setup_fields(thd, 0, update_fields, MARK_COLUMNS_WRITE, 0, 0))
322 323
    return -1;

324 325 326 327
  if (insert_table_list->effective_algorithm == VIEW_ALGORITHM_MERGE &&
      check_view_single_update(update_fields, insert_table_list, map))
    return -1;

328 329 330
  if (table->timestamp_field)
  {
    /* Don't set timestamp column if this is modified. */
331 332
    if (bitmap_is_set(table->write_set,
                      table->timestamp_field->field_index))
333 334
      clear_timestamp_auto_bits(table->timestamp_field_type,
                                TIMESTAMP_AUTO_SET_ON_UPDATE);
335 336 337
    if (timestamp_mark)
      bitmap_set_bit(table->write_set,
                     table->timestamp_field->field_index);
338 339 340 341
  }
  return 0;
}

unknown's avatar
unknown committed
342
/*
343 344 345 346 347 348 349 350 351 352 353 354
  Prepare triggers  for INSERT-like statement.

  SYNOPSIS
    prepare_triggers_for_insert_stmt()
      table   Table to which insert will happen

  NOTE
    Prepare triggers for INSERT-like statement by marking fields
    used by triggers and inform handlers that batching of UPDATE/DELETE 
    cannot be done if there are BEFORE UPDATE/DELETE triggers.
*/

unknown's avatar
unknown committed
355
void prepare_triggers_for_insert_stmt(TABLE *table)
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379
{
  if (table->triggers)
  {
    if (table->triggers->has_triggers(TRG_EVENT_DELETE,
                                      TRG_ACTION_AFTER))
    {
      /*
        The table has AFTER DELETE triggers that might access to 
        subject table and therefore might need delete to be done 
        immediately. So we turn-off the batching.
      */ 
      (void) table->file->extra(HA_EXTRA_DELETE_CANNOT_BATCH);
    }
    if (table->triggers->has_triggers(TRG_EVENT_UPDATE,
                                      TRG_ACTION_AFTER))
    {
      /*
        The table has AFTER UPDATE triggers that might access to subject 
        table and therefore might need update to be done immediately. 
        So we turn-off the batching.
      */ 
      (void) table->file->extra(HA_EXTRA_UPDATE_CANNOT_BATCH);
    }
  }
unknown's avatar
unknown committed
380
  table->mark_columns_needed_for_insert();
381 382
}

383

384 385 386 387 388 389 390 391
/**
  Upgrade table-level lock of INSERT statement to TL_WRITE if
  a more concurrent lock is infeasible for some reason. This is
  necessary for engines without internal locking support (MyISAM).
  An engine with internal locking implementation might later
  downgrade the lock in handler::store_lock() method.
*/

unknown's avatar
unknown committed
392
static
393 394 395 396 397 398 399 400 401 402 403 404 405 406 407
void upgrade_lock_type(THD *thd, thr_lock_type *lock_type,
                       enum_duplicates duplic,
                       bool is_multi_insert)
{
  if (duplic == DUP_UPDATE ||
      duplic == DUP_REPLACE && *lock_type == TL_WRITE_CONCURRENT_INSERT)
  {
    *lock_type= TL_WRITE;
    return;
  }

  if (*lock_type == TL_WRITE_DELAYED)
  {
    /*
      We do not use delayed threads if:
unknown's avatar
unknown committed
408 409 410 411 412
      - we're running in the safe mode or skip-new mode -- the
        feature is disabled in these modes
      - we're executing this statement on a replication slave --
        we need to ensure serial execution of queries on the
        slave
413 414
      - it is INSERT .. ON DUPLICATE KEY UPDATE - in this case the
        insert cannot be concurrent
unknown's avatar
unknown committed
415 416 417 418 419 420 421 422 423 424 425 426
      - this statement is directly or indirectly invoked from
        a stored function or trigger (under pre-locking) - to
        avoid deadlocks, since INSERT DELAYED involves a lock
        upgrade (TL_WRITE_DELAYED -> TL_WRITE) which we should not
        attempt while keeping other table level locks.
      - this statement itself may require pre-locking.
        We should upgrade the lock even though in most cases
        delayed functionality may work. Unfortunately, we can't
        easily identify whether the subject table is not used in
        the statement indirectly via a stored function or trigger:
        if it is used, that will lead to a deadlock between the
        client connection and the delayed thread.
427 428 429
    */
    if (specialflag & (SPECIAL_NO_NEW_FUNC | SPECIAL_SAFE_MODE) ||
        thd->slave_thread ||
unknown's avatar
unknown committed
430 431 432
        thd->variables.max_insert_delayed_threads == 0 ||
        thd->prelocked_mode ||
        thd->lex->uses_stored_routines())
433 434 435 436 437 438
    {
      *lock_type= TL_WRITE;
      return;
    }
    bool log_on= (thd->options & OPTION_BIN_LOG ||
                  ! (thd->security_ctx->master_access & SUPER_ACL));
439 440
    if (global_system_variables.binlog_format == BINLOG_FORMAT_STMT &&
        log_on && mysql_bin_log.is_open() && is_multi_insert)
441 442 443 444 445 446 447 448 449 450 451 452 453 454
    {
      /*
        Statement-based binary logging does not work in this case, because:
        a) two concurrent statements may have their rows intermixed in the
        queue, leading to autoincrement replication problems on slave (because
        the values generated used for one statement don't depend only on the
        value generated for the first row of this statement, so are not
        replicable)
        b) if first row of the statement has an error the full statement is
        not binlogged, while next rows of the statement may be inserted.
        c) if first row succeeds, statement is binlogged immediately with a
        zero error code (i.e. "no error"), if then second row fails, query
        will fail on slave too and slave will stop (wrongly believing that the
        master got no error).
455 456 457 458 459 460 461 462 463
        So we fallback to non-delayed INSERT.
        Note that to be fully correct, we should test the "binlog format which
        the delayed thread is going to use for this row". But in the common case
        where the global binlog format is not changed and the session binlog
        format may be changed, that is equal to the global binlog format.
        We test it without mutex for speed reasons (condition rarely true), and
        in the common case (global not changed) it is as good as without mutex;
        if global value is changed, anyway there is uncertainty as the delayed
        thread may be old and use the before-the-change value.
464 465 466 467 468 469 470
      */
      *lock_type= TL_WRITE;
    }
  }
}


unknown's avatar
unknown committed
471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536
/**
  Find or create a delayed insert thread for the first table in
  the table list, then open and lock the remaining tables.
  If a table can not be used with insert delayed, upgrade the lock
  and open and lock all tables using the standard mechanism.

  @param thd         thread context
  @param table_list  list of "descriptors" for tables referenced
                     directly in statement SQL text.
                     The first element in the list corresponds to
                     the destination table for inserts, remaining
                     tables, if any, are usually tables referenced
                     by sub-queries in the right part of the
                     INSERT.

  @return Status of the operation. In case of success 'table'
  member of every table_list element points to an instance of
  class TABLE.

  @sa open_and_lock_tables for more information about MySQL table
  level locking
*/

static
bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
{
  DBUG_ENTER("open_and_lock_for_insert_delayed");

#ifndef EMBEDDED_LIBRARY
  if (delayed_get_table(thd, table_list))
    DBUG_RETURN(TRUE);

  if (table_list->table)
  {
    /*
      Open tables used for sub-selects or in stored functions, will also
      cache these functions.
    */
    if (open_and_lock_tables(thd, table_list->next_global))
    {
      end_delayed_insert(thd);
      DBUG_RETURN(TRUE);
    }
    /*
      First table was not processed by open_and_lock_tables(),
      we need to set updatability flag "by hand".
    */
    if (!table_list->derived && !table_list->view)
      table_list->updatable= 1;  // usual table
    DBUG_RETURN(FALSE);
  }
#endif
  /*
    * This is embedded library and we don't have auxiliary
    threads OR
    * a lock upgrade was requested inside delayed_get_table
      because
      - there are too many delayed insert threads OR
      - the table has triggers.
    Use a normal insert.
  */
  table_list->lock_type= TL_WRITE;
  DBUG_RETURN(open_and_lock_tables(thd, table_list));
}


537 538 539 540
/**
  INSERT statement implementation
*/

unknown's avatar
unknown committed
541 542 543 544 545
bool mysql_insert(THD *thd,TABLE_LIST *table_list,
                  List<Item> &fields,
                  List<List_item> &values_list,
                  List<Item> &update_fields,
                  List<Item> &update_values,
546 547
                  enum_duplicates duplic,
		  bool ignore)
unknown's avatar
unknown committed
548
{
549
  int error, res;
550
  bool transactional_table, joins_freed= FALSE;
551
  bool changed;
unknown's avatar
unknown committed
552 553 554 555
  uint value_count;
  ulong counter = 1;
  ulonglong id;
  COPY_INFO info;
556
  TABLE *table= 0;
unknown's avatar
unknown committed
557
  List_iterator_fast<List_item> its(values_list);
unknown's avatar
unknown committed
558
  List_item *values;
unknown's avatar
unknown committed
559
  Name_resolution_context *context;
560
  Name_resolution_context_state ctx_state;
561
#ifndef EMBEDDED_LIBRARY
unknown's avatar
unknown committed
562
  char *query= thd->query;
unknown's avatar
unknown committed
563 564 565 566 567 568
  /*
    log_on is about delayed inserts only.
    By default, both logs are enabled (this won't cause problems if the server
    runs without --log-update or --log-bin).
  */
  bool log_on= ((thd->options & OPTION_BIN_LOG) ||
unknown's avatar
unknown committed
569
                (!(thd->security_ctx->master_access & SUPER_ACL)));
unknown's avatar
unknown committed
570
#endif
unknown's avatar
unknown committed
571
  thr_lock_type lock_type = table_list->lock_type;
unknown's avatar
unknown committed
572
  Item *unused_conds= 0;
unknown's avatar
unknown committed
573 574
  DBUG_ENTER("mysql_insert");

575
  /*
576 577 578 579 580 581 582 583 584 585 586 587
    Upgrade lock type if the requested lock is incompatible with
    the current connection mode or table operation.
  */
  upgrade_lock_type(thd, &table_list->lock_type, duplic,
                    values_list.elements > 1);

  /*
    We can't write-delayed into a table locked with LOCK TABLES:
    this will lead to a deadlock, since the delayed thread will
    never be able to get a lock on the table. QQQ: why not
    upgrade the lock here instead?
  */
unknown's avatar
unknown committed
588
  if (table_list->lock_type == TL_WRITE_DELAYED && thd->locked_tables &&
589
      find_locked_table(thd, table_list->db, table_list->table_name))
590
  {
591 592 593
    my_error(ER_DELAYED_INSERT_TABLE_LOCKED, MYF(0),
             table_list->table_name);
    DBUG_RETURN(TRUE);
594
  }
unknown's avatar
unknown committed
595

unknown's avatar
unknown committed
596
  if (table_list->lock_type == TL_WRITE_DELAYED)
unknown's avatar
unknown committed
597
  {
unknown's avatar
unknown committed
598 599
    if (open_and_lock_for_insert_delayed(thd, table_list))
      DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
600 601
  }
  else
unknown's avatar
unknown committed
602 603 604 605
  {
    if (open_and_lock_tables(thd, table_list))
      DBUG_RETURN(TRUE);
  }
606

unknown's avatar
unknown committed
607
  thd->proc_info="init";
608
  thd->used_tables=0;
unknown's avatar
unknown committed
609
  values= its++;
unknown's avatar
unknown committed
610
  value_count= values->elements;
unknown's avatar
unknown committed
611

unknown's avatar
VIEW  
unknown committed
612
  if (mysql_prepare_insert(thd, table_list, table, fields, values,
unknown's avatar
unknown committed
613
			   update_fields, update_values, duplic, &unused_conds,
unknown's avatar
unknown committed
614 615 616 617 618
                           FALSE,
                           (fields.elements || !value_count),
                           !ignore && (thd->variables.sql_mode &
                                       (MODE_STRICT_TRANS_TABLES |
                                        MODE_STRICT_ALL_TABLES))))
unknown's avatar
unknown committed
619
    goto abort;
620

621 622
  /* mysql_prepare_insert set table_list->table if it was not set */
  table= table_list->table;
unknown's avatar
unknown committed
623
  lock_type= table_list->lock_type;
624

unknown's avatar
unknown committed
625
  context= &thd->lex->select_lex.context;
626 627 628 629 630 631 632 633 634
  /*
    These three asserts test the hypothesis that the resetting of the name
    resolution context below is not necessary at all since the list of local
    tables for INSERT always consists of one table.
  */
  DBUG_ASSERT(!table_list->next_local);
  DBUG_ASSERT(!context->table_list->next_local);
  DBUG_ASSERT(!context->first_name_resolution_table->next_name_resolution_table);

unknown's avatar
unknown committed
635
  /* Save the state of the current name resolution context. */
636
  ctx_state.save_state(context, table_list);
unknown's avatar
unknown committed
637 638 639 640 641

  /*
    Perform name resolution only in the first table - 'table_list',
    which is the table that is inserted into.
  */
642
  table_list->next_local= 0;
unknown's avatar
unknown committed
643 644
  context->resolve_in_table_list_only(table_list);

645
  while ((values= its++))
unknown's avatar
unknown committed
646 647 648 649
  {
    counter++;
    if (values->elements != value_count)
    {
650
      my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), counter);
unknown's avatar
unknown committed
651 652
      goto abort;
    }
653
    if (setup_fields(thd, 0, *values, MARK_COLUMNS_READ, 0, 0))
unknown's avatar
unknown committed
654 655 656
      goto abort;
  }
  its.rewind ();
unknown's avatar
unknown committed
657 658
 
  /* Restore the current context. */
659
  ctx_state.restore_state(context, table_list);
unknown's avatar
unknown committed
660

unknown's avatar
unknown committed
661
  /*
unknown's avatar
unknown committed
662
    Fill in the given fields and dump it to the table file
unknown's avatar
unknown committed
663
  */
unknown's avatar
unknown committed
664
  info.records= info.deleted= info.copied= info.updated= 0;
665
  info.ignore= ignore;
unknown's avatar
unknown committed
666
  info.handle_duplicates=duplic;
667 668
  info.update_fields= &update_fields;
  info.update_values= &update_values;
unknown's avatar
unknown committed
669
  info.view= (table_list->view ? table_list : 0);
670

671 672 673
  /*
    Count warnings for all inserts.
    For single line insert, generate an error if try to set a NOT NULL field
unknown's avatar
unknown committed
674
    to NULL.
675
  */
unknown's avatar
unknown committed
676
  thd->count_cuted_fields= ((values_list.elements == 1 &&
unknown's avatar
unknown committed
677
                             !ignore) ?
678 679
			    CHECK_FIELD_ERROR_FOR_NULL :
			    CHECK_FIELD_WARN);
unknown's avatar
unknown committed
680 681 682
  thd->cuted_fields = 0L;
  table->next_number_field=table->found_next_number_field;

683 684 685 686 687 688 689 690
#ifdef HAVE_REPLICATION
  if (thd->slave_thread &&
      (info.handle_duplicates == DUP_UPDATE) &&
      (table->next_number_field != NULL) &&
      rpl_master_has_bug(&active_mi->rli, 24432))
    goto abort;
#endif

unknown's avatar
unknown committed
691 692
  error=0;
  thd->proc_info="update";
unknown's avatar
unknown committed
693
  if (duplic != DUP_ERROR || ignore)
unknown's avatar
unknown committed
694
    table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
695 696 697
  if (duplic == DUP_REPLACE &&
      (!table->triggers || !table->triggers->has_delete_triggers()))
    table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
698 699 700 701
  /*
    let's *try* to start bulk inserts. It won't necessary
    start them as values_list.elements should be greater than
    some - handler dependent - threshold.
702 703 704 705
    We should not start bulk inserts if this statement uses
    functions or invokes triggers since they may access
    to the same table and therefore should not see its
    inconsistent state created by this optimization.
706 707 708 709
    So we call start_bulk_insert to perform nesessary checks on
    values_list.elements, and - if nothing else - to initialize
    the code to make the call of end_bulk_insert() below safe.
  */
710
  if (lock_type != TL_WRITE_DELAYED && !thd->prelocked_mode)
711
    table->file->ha_start_bulk_insert(values_list.elements);
712

713
  thd->no_trans_update.stmt= FALSE;
unknown's avatar
unknown committed
714 715 716
  thd->abort_on_warning= (!ignore && (thd->variables.sql_mode &
                                       (MODE_STRICT_TRANS_TABLES |
                                        MODE_STRICT_ALL_TABLES)));
unknown's avatar
unknown committed
717

unknown's avatar
unknown committed
718 719
  prepare_triggers_for_insert_stmt(table);

720

721 722 723 724
  if (table_list->prepare_where(thd, 0, TRUE) ||
      table_list->prepare_check_option(thd))
    error= 1;

unknown's avatar
unknown committed
725
  while ((values= its++))
unknown's avatar
unknown committed
726 727 728
  {
    if (fields.elements || !value_count)
    {
729
      restore_record(table,s->default_values);	// Get empty record
unknown's avatar
unknown committed
730 731 732
      if (fill_record_n_invoke_before_triggers(thd, fields, *values, 0,
                                               table->triggers,
                                               TRG_EVENT_INSERT))
unknown's avatar
unknown committed
733
      {
unknown's avatar
unknown committed
734
	if (values_list.elements != 1 && !thd->net.report_error)
unknown's avatar
unknown committed
735 736 737 738
	{
	  info.records++;
	  continue;
	}
unknown's avatar
unknown committed
739 740 741 742 743
	/*
	  TODO: set thd->abort_on_warning if values_list.elements == 1
	  and check that all items return warning in case of problem with
	  storing field.
        */
unknown's avatar
unknown committed
744 745 746 747 748 749
	error=1;
	break;
      }
    }
    else
    {
750
      if (thd->used_tables)			// Column used in values()
751
	restore_record(table,s->default_values);	// Get empty record
752
      else
unknown's avatar
unknown committed
753 754 755 756 757 758 759 760 761 762 763
      {
        /*
          Fix delete marker. No need to restore rest of record since it will
          be overwritten by fill_record() anyway (and fill_record() does not
          use default values in this case).
        */
	table->record[0][0]= table->s->default_values[0];
      }
      if (fill_record_n_invoke_before_triggers(thd, table->field, *values, 0,
                                               table->triggers,
                                               TRG_EVENT_INSERT))
unknown's avatar
unknown committed
764
      {
unknown's avatar
unknown committed
765
	if (values_list.elements != 1 && ! thd->net.report_error)
unknown's avatar
unknown committed
766 767 768 769 770 771 772 773
	{
	  info.records++;
	  continue;
	}
	error=1;
	break;
      }
    }
774

775 776 777
    if ((res= table_list->view_check_option(thd,
					    (values_list.elements == 1 ?
					     0 :
778
					     ignore))) ==
unknown's avatar
unknown committed
779 780 781
        VIEW_CHECK_SKIP)
      continue;
    else if (res == VIEW_CHECK_ERROR)
unknown's avatar
unknown committed
782
    {
unknown's avatar
unknown committed
783 784
      error= 1;
      break;
unknown's avatar
unknown committed
785
    }
786
#ifndef EMBEDDED_LIBRARY
unknown's avatar
unknown committed
787
    if (lock_type == TL_WRITE_DELAYED)
unknown's avatar
unknown committed
788
    {
789 790
      LEX_STRING const st_query = { query, thd->query_length };
      error=write_delayed(thd, table, duplic, st_query, ignore, log_on);
unknown's avatar
unknown committed
791 792 793
      query=0;
    }
    else
794
#endif
unknown's avatar
unknown committed
795
      error=write_record(thd, table ,&info);
796 797
    if (error)
      break;
798
    thd->row_count++;
unknown's avatar
unknown committed
799
  }
800

801 802 803
  free_underlaid_joins(thd, &thd->lex->select_lex);
  joins_freed= TRUE;

804 805 806 807
  /*
    Now all rows are inserted.  Time to update logs and sends response to
    user
  */
808
#ifndef EMBEDDED_LIBRARY
unknown's avatar
unknown committed
809 810
  if (lock_type == TL_WRITE_DELAYED)
  {
811 812 813 814 815
    if (!error)
    {
      info.copied=values_list.elements;
      end_delayed_insert(thd);
    }
unknown's avatar
unknown committed
816
    query_cache_invalidate3(thd, table_list, 1);
unknown's avatar
unknown committed
817 818
  }
  else
819
#endif
unknown's avatar
unknown committed
820
  {
821 822 823 824 825
    /*
      Do not do this release if this is a delayed insert, it would steal
      auto_inc values from the delayed_insert thread as they share TABLE.
    */
    table->file->ha_release_auto_increment();
826
    if (!thd->prelocked_mode && table->file->ha_end_bulk_insert() && !error)
827
    {
unknown's avatar
unknown committed
828 829
      table->file->print_error(my_errno,MYF(0));
      error=1;
830
    }
831
    transactional_table= table->file->has_transactions();
unknown's avatar
unknown committed
832

833
    if ((changed= (info.copied || info.deleted || info.updated)))
unknown's avatar
unknown committed
834
    {
835 836 837 838 839 840 841
      /*
        Invalidate the table in the query cache if something changed.
        For the transactional algorithm to work the invalidation must be
        before binlog writing and ha_autocommit_or_rollback
      */
      query_cache_invalidate3(thd, table_list, 1);
      if (error <= 0 || !transactional_table)
842
      {
843
        if (mysql_bin_log.is_open())
844
        {
845
          if (error <= 0)
846 847 848 849 850 851 852 853 854 855 856
          {
            /*
              [Guilhem wrote] Temporary errors may have filled
              thd->net.last_error/errno.  For example if there has
              been a disk full error when writing the row, and it was
              MyISAM, then thd->net.last_error/errno will be set to
              "disk full"... and the my_pwrite() will wait until free
              space appears, and so when it finishes then the
              write_row() was entirely successful
            */
            /* todo: consider removing */
857
            thd->clear_error();
858
          }
859 860
          /* bug#22725:

861 862 863 864 865 866 867 868 869 870
          A query which per-row-loop can not be interrupted with
          KILLED, like INSERT, and that does not invoke stored
          routines can be binlogged with neglecting the KILLED error.
          
          If there was no error (error == zero) until after the end of
          inserting loop the KILLED flag that appeared later can be
          disregarded since previously possible invocation of stored
          routines did not result in any error due to the KILLED.  In
          such case the flag is ignored for constructing binlog event.
          */
871
          DBUG_ASSERT(thd->killed != THD::KILL_BAD_DATA || error > 0);
unknown's avatar
unknown committed
872 873
          if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                                thd->query, thd->query_length,
874 875
                                transactional_table, FALSE,
                                (error>0) ? thd->killed : THD::NOT_KILLED) &&
unknown's avatar
unknown committed
876 877
              transactional_table)
          {
878
            error=1;
unknown's avatar
unknown committed
879
          }
880
        }
881
        if (!transactional_table)
unknown's avatar
unknown committed
882
          thd->no_trans_update.all= TRUE;
883
      }
unknown's avatar
unknown committed
884
    }
885
    if (transactional_table)
886
      error=ha_autocommit_or_rollback(thd,error);
887
    
unknown's avatar
unknown committed
888 889 890
    if (thd->lock)
    {
      mysql_unlock_tables(thd, thd->lock);
891 892 893 894 895 896 897 898 899 900
      /*
        Invalidate the table in the query cache if something changed
        after unlocking when changes become fisible.
        TODO: this is workaround. right way will be move invalidating in
        the unlock procedure.
      */
      if (lock_type ==  TL_WRITE_CONCURRENT_INSERT && changed)
      {
        query_cache_invalidate3(thd, table_list, 1);
      }
unknown's avatar
unknown committed
901 902 903 904
      thd->lock=0;
    }
  }
  thd->proc_info="end";
905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920
  /*
    We'll report to the client this id:
    - if the table contains an autoincrement column and we successfully
    inserted an autogenerated value, the autogenerated value.
    - if the table contains no autoincrement column and LAST_INSERT_ID(X) was
    called, X.
    - if the table contains an autoincrement column, and some rows were
    inserted, the id of the last "inserted" row (if IGNORE, that value may not
    have been really inserted but ignored).
  */
  id= (thd->first_successful_insert_id_in_cur_stmt > 0) ?
    thd->first_successful_insert_id_in_cur_stmt :
    (thd->arg_of_last_insert_id_function ?
     thd->first_successful_insert_id_in_prev_stmt :
     ((table->next_number_field && info.copied) ?
     table->next_number_field->val_int() : 0));
unknown's avatar
unknown committed
921
  table->next_number_field=0;
922
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
923
  table->auto_increment_field_not_null= FALSE;
unknown's avatar
unknown committed
924
  if (duplic != DUP_ERROR || ignore)
unknown's avatar
unknown committed
925
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
926 927 928
  if (duplic == DUP_REPLACE &&
      (!table->triggers || !table->triggers->has_delete_triggers()))
    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
929

unknown's avatar
unknown committed
930 931 932 933
  if (error)
    goto abort;
  if (values_list.elements == 1 && (!(thd->options & OPTION_WARNINGS) ||
				    !thd->cuted_fields))
934
  {
935 936 937
    thd->row_count_func= info.copied + info.deleted +
                         ((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
                          info.touched : info.updated);
938
    send_ok(thd, (ulong) thd->row_count_func, id);
939
  }
940 941
  else
  {
unknown's avatar
unknown committed
942
    char buff[160];
943 944
    ha_rows updated=((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
                     info.touched : info.updated);
945
    if (ignore)
946 947 948
      sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
	      (lock_type == TL_WRITE_DELAYED) ? (ulong) 0 :
	      (ulong) (info.records - info.copied), (ulong) thd->cuted_fields);
unknown's avatar
unknown committed
949
    else
950
      sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
951 952
	      (ulong) (info.deleted + updated), (ulong) thd->cuted_fields);
    thd->row_count_func= info.copied + info.deleted + updated;
953
    ::send_ok(thd, (ulong) thd->row_count_func, id, buff);
unknown's avatar
unknown committed
954
  }
955
  thd->abort_on_warning= 0;
unknown's avatar
unknown committed
956
  DBUG_RETURN(FALSE);
unknown's avatar
unknown committed
957 958

abort:
959
#ifndef EMBEDDED_LIBRARY
unknown's avatar
unknown committed
960 961
  if (lock_type == TL_WRITE_DELAYED)
    end_delayed_insert(thd);
962
#endif
963
  if (table != NULL)
964
    table->file->ha_release_auto_increment();
965 966
  if (!joins_freed)
    free_underlaid_joins(thd, &thd->lex->select_lex);
unknown's avatar
unknown committed
967
  thd->abort_on_warning= 0;
unknown's avatar
unknown committed
968
  DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
969 970 971
}


unknown's avatar
VIEW  
unknown committed
972 973 974 975 976
/*
  Additional check for insertability for VIEW

  SYNOPSIS
    check_view_insertability()
977
    thd     - thread handler
unknown's avatar
VIEW  
unknown committed
978 979
    view    - reference on VIEW

980 981 982 983 984 985
  IMPLEMENTATION
    A view is insertable if the folloings are true:
    - All columns in the view are columns from a table
    - All not used columns in table have a default values
    - All field in view are unique (not referring to the same column)

unknown's avatar
VIEW  
unknown committed
986 987
  RETURN
    FALSE - OK
988 989 990
      view->contain_auto_increment is 1 if and only if the view contains an
      auto_increment field

unknown's avatar
VIEW  
unknown committed
991 992 993
    TRUE  - can't be used for insert
*/

994
static bool check_view_insertability(THD * thd, TABLE_LIST *view)
unknown's avatar
VIEW  
unknown committed
995
{
996
  uint num= view->view->select_lex.item_list.elements;
unknown's avatar
VIEW  
unknown committed
997
  TABLE *table= view->table;
998 999 1000
  Field_translator *trans_start= view->field_translation,
		   *trans_end= trans_start + num;
  Field_translator *trans;
unknown's avatar
unknown committed
1001
  uint used_fields_buff_size= bitmap_buffer_size(table->s->fields);
unknown's avatar
unknown committed
1002
  uint32 *used_fields_buff= (uint32*)thd->alloc(used_fields_buff_size);
1003
  MY_BITMAP used_fields;
unknown's avatar
unknown committed
1004
  enum_mark_columns save_mark_used_columns= thd->mark_used_columns;
1005 1006
  DBUG_ENTER("check_key_in_view");

1007 1008 1009
  if (!used_fields_buff)
    DBUG_RETURN(TRUE);  // EOM

unknown's avatar
VIEW  
unknown committed
1010 1011
  DBUG_ASSERT(view->table != 0 && view->field_translation != 0);

1012
  VOID(bitmap_init(&used_fields, used_fields_buff, table->s->fields, 0));
1013 1014
  bitmap_clear_all(&used_fields);

unknown's avatar
VIEW  
unknown committed
1015
  view->contain_auto_increment= 0;
1016 1017 1018 1019
  /* 
    we must not set query_id for fields as they're not 
    really used in this context
  */
unknown's avatar
unknown committed
1020
  thd->mark_used_columns= MARK_COLUMNS_NONE;
unknown's avatar
VIEW  
unknown committed
1021
  /* check simplicity and prepare unique test of view */
1022
  for (trans= trans_start; trans != trans_end; trans++)
unknown's avatar
VIEW  
unknown committed
1023
  {
1024
    if (!trans->item->fixed && trans->item->fix_fields(thd, &trans->item))
1025
    {
unknown's avatar
unknown committed
1026
      thd->mark_used_columns= save_mark_used_columns;
1027 1028
      DBUG_RETURN(TRUE);
    }
1029
    Item_field *field;
unknown's avatar
VIEW  
unknown committed
1030
    /* simple SELECT list entry (field without expression) */
unknown's avatar
merge  
unknown committed
1031
    if (!(field= trans->item->filed_for_view_update()))
1032
    {
unknown's avatar
unknown committed
1033
      thd->mark_used_columns= save_mark_used_columns;
unknown's avatar
VIEW  
unknown committed
1034
      DBUG_RETURN(TRUE);
1035
    }
1036
    if (field->field->unireg_check == Field::NEXT_NUMBER)
unknown's avatar
VIEW  
unknown committed
1037 1038
      view->contain_auto_increment= 1;
    /* prepare unique test */
unknown's avatar
unknown committed
1039 1040 1041 1042 1043
    /*
      remove collation (or other transparent for update function) if we have
      it
    */
    trans->item= field;
unknown's avatar
VIEW  
unknown committed
1044
  }
unknown's avatar
unknown committed
1045
  thd->mark_used_columns= save_mark_used_columns;
unknown's avatar
VIEW  
unknown committed
1046
  /* unique test */
1047
  for (trans= trans_start; trans != trans_end; trans++)
unknown's avatar
VIEW  
unknown committed
1048
  {
1049
    /* Thanks to test above, we know that all columns are of type Item_field */
1050
    Item_field *field= (Item_field *)trans->item;
1051 1052 1053
    /* check fields belong to table in which we are inserting */
    if (field->field->table == table &&
        bitmap_fast_test_and_set(&used_fields, field->field->field_index))
unknown's avatar
VIEW  
unknown committed
1054 1055 1056 1057 1058 1059 1060
      DBUG_RETURN(TRUE);
  }

  DBUG_RETURN(FALSE);
}


unknown's avatar
unknown committed
1061
/*
1062
  Check if table can be updated
unknown's avatar
unknown committed
1063 1064

  SYNOPSIS
1065 1066
     mysql_prepare_insert_check_table()
     thd		Thread handle
unknown's avatar
unknown committed
1067
     table_list		Table list
1068 1069
     fields		List of fields to be updated
     where		Pointer to where clause
unknown's avatar
unknown committed
1070
     select_insert      Check is making for SELECT ... INSERT
1071 1072

   RETURN
unknown's avatar
unknown committed
1073 1074
     FALSE ok
     TRUE  ERROR
unknown's avatar
unknown committed
1075
*/
1076

unknown's avatar
unknown committed
1077
static bool mysql_prepare_insert_check_table(THD *thd, TABLE_LIST *table_list,
1078
                                             List<Item> &fields,
unknown's avatar
unknown committed
1079
                                             bool select_insert)
unknown's avatar
unknown committed
1080
{
unknown's avatar
VIEW  
unknown committed
1081
  bool insert_into_view= (table_list->view != 0);
1082
  DBUG_ENTER("mysql_prepare_insert_check_table");
unknown's avatar
VIEW  
unknown committed
1083

1084 1085 1086 1087 1088 1089 1090
  /*
     first table in list is the one we'll INSERT into, requires INSERT_ACL.
     all others require SELECT_ACL only. the ACL requirement below is for
     new leaves only anyway (view-constituents), so check for SELECT rather
     than INSERT.
  */

1091 1092
  if (setup_tables_and_check_access(thd, &thd->lex->select_lex.context,
                                    &thd->lex->select_lex.top_join_list,
1093
                                    table_list,
1094
                                    &thd->lex->select_lex.leaf_tables,
1095
                                    select_insert, INSERT_ACL, SELECT_ACL))
unknown's avatar
unknown committed
1096
    DBUG_RETURN(TRUE);
unknown's avatar
VIEW  
unknown committed
1097 1098 1099 1100

  if (insert_into_view && !fields.elements)
  {
    thd->lex->empty_field_list_on_rset= 1;
1101 1102 1103 1104
    if (!table_list->table)
    {
      my_error(ER_VIEW_NO_INSERT_FIELD_LIST, MYF(0),
               table_list->view_db.str, table_list->view_name.str);
unknown's avatar
unknown committed
1105
      DBUG_RETURN(TRUE);
1106
    }
1107
    DBUG_RETURN(insert_view_fields(thd, &fields, table_list));
unknown's avatar
VIEW  
unknown committed
1108 1109
  }

unknown's avatar
unknown committed
1110
  DBUG_RETURN(FALSE);
1111 1112 1113 1114 1115 1116 1117 1118 1119 1120
}


/*
  Prepare items in INSERT statement

  SYNOPSIS
    mysql_prepare_insert()
    thd			Thread handler
    table_list	        Global/local table list
unknown's avatar
unknown committed
1121 1122
    table		Table to insert into (can be NULL if table should
			be taken from table_list->table)    
unknown's avatar
unknown committed
1123 1124
    where		Where clause (for insert ... select)
    select_insert	TRUE if INSERT ... SELECT statement
unknown's avatar
unknown committed
1125 1126 1127 1128
    check_fields        TRUE if need to check that all INSERT fields are 
                        given values.
    abort_on_warning    whether to report if some INSERT field is not 
                        assigned as an error (TRUE) or as a warning (FALSE).
1129

unknown's avatar
unknown committed
1130 1131 1132 1133 1134
  TODO (in far future)
    In cases of:
    INSERT INTO t1 SELECT a, sum(a) as sum1 from t2 GROUP BY a
    ON DUPLICATE KEY ...
    we should be able to refer to sum1 in the ON DUPLICATE KEY part
unknown's avatar
unknown committed
1135

1136 1137 1138
  WARNING
    You MUST set table->insert_values to 0 after calling this function
    before releasing the table object.
unknown's avatar
unknown committed
1139
  
1140
  RETURN VALUE
unknown's avatar
unknown committed
1141 1142
    FALSE OK
    TRUE  error
1143 1144
*/

unknown's avatar
unknown committed
1145
bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
unknown's avatar
unknown committed
1146
                          TABLE *table, List<Item> &fields, List_item *values,
unknown's avatar
unknown committed
1147
                          List<Item> &update_fields, List<Item> &update_values,
unknown's avatar
unknown committed
1148
                          enum_duplicates duplic,
unknown's avatar
unknown committed
1149 1150
                          COND **where, bool select_insert,
                          bool check_fields, bool abort_on_warning)
1151
{
unknown's avatar
unknown committed
1152
  SELECT_LEX *select_lex= &thd->lex->select_lex;
unknown's avatar
unknown committed
1153
  Name_resolution_context *context= &select_lex->context;
1154
  Name_resolution_context_state ctx_state;
1155
  bool insert_into_view= (table_list->view != 0);
unknown's avatar
unknown committed
1156
  bool res= 0;
1157
  table_map map= 0;
1158
  DBUG_ENTER("mysql_prepare_insert");
unknown's avatar
unknown committed
1159 1160 1161
  DBUG_PRINT("enter", ("table_list 0x%lx, table 0x%lx, view %d",
		       (ulong)table_list, (ulong)table,
		       (int)insert_into_view));
1162 1163
  /* INSERT should have a SELECT or VALUES clause */
  DBUG_ASSERT (!select_insert || !values);
unknown's avatar
unknown committed
1164

1165 1166 1167 1168 1169 1170 1171
  /*
    For subqueries in VALUES() we should not see the table in which we are
    inserting (for INSERT ... SELECT this is done by changing table_list,
    because INSERT ... SELECT share SELECT_LEX it with SELECT.
  */
  if (!select_insert)
  {
unknown's avatar
unknown committed
1172
    for (SELECT_LEX_UNIT *un= select_lex->first_inner_unit();
1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
         un;
         un= un->next_unit())
    {
      for (SELECT_LEX *sl= un->first_select();
           sl;
           sl= sl->next_select())
      {
        sl->context.outer_context= 0;
      }
    }
  }

unknown's avatar
unknown committed
1185
  if (duplic == DUP_UPDATE)
unknown's avatar
unknown committed
1186 1187
  {
    /* it should be allocated before Item::fix_fields() */
unknown's avatar
unknown committed
1188
    if (table_list->set_insert_values(thd->mem_root))
unknown's avatar
unknown committed
1189
      DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1190
  }
unknown's avatar
unknown committed
1191

1192
  if (mysql_prepare_insert_check_table(thd, table_list, fields, select_insert))
unknown's avatar
unknown committed
1193
    DBUG_RETURN(TRUE);
unknown's avatar
VIEW  
unknown committed
1194

unknown's avatar
unknown committed
1195 1196

  /* Prepare the fields in the statement. */
1197
  if (values)
unknown's avatar
unknown committed
1198
  {
1199 1200 1201 1202 1203 1204
    /* if we have INSERT ... VALUES () we cannot have a GROUP BY clause */
    DBUG_ASSERT (!select_lex->group_list.elements);

    /* Save the state of the current name resolution context. */
    ctx_state.save_state(context, table_list);

1205
    /*
1206 1207 1208 1209 1210 1211
      Perform name resolution only in the first table - 'table_list',
      which is the table that is inserted into.
     */
    table_list->next_local= 0;
    context->resolve_in_table_list_only(table_list);

unknown's avatar
unknown committed
1212 1213
    res= check_insert_fields(thd, context->table_list, fields, *values,
                             !insert_into_view, &map) ||
unknown's avatar
unknown committed
1214
      setup_fields(thd, 0, *values, MARK_COLUMNS_READ, 0, 0);
unknown's avatar
unknown committed
1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227

    if (!res && check_fields)
    {
      bool saved_abort_on_warning= thd->abort_on_warning;
      thd->abort_on_warning= abort_on_warning;
      res= check_that_all_fields_are_given_values(thd, 
                                                  table ? table : 
                                                  context->table_list->table,
                                                  context->table_list);
      thd->abort_on_warning= saved_abort_on_warning;
    }

    if (!res && duplic == DUP_UPDATE)
unknown's avatar
unknown committed
1228
    {
1229 1230 1231
      select_lex->no_wrap_view_item= TRUE;
      res= check_update_fields(thd, context->table_list, update_fields, &map);
      select_lex->no_wrap_view_item= FALSE;
unknown's avatar
unknown committed
1232
    }
1233 1234 1235 1236

    /* Restore the current context. */
    ctx_state.restore_state(context, table_list);

unknown's avatar
unknown committed
1237
    if (!res)
1238
      res= setup_fields(thd, 0, update_values, MARK_COLUMNS_READ, 0, 0);
unknown's avatar
unknown committed
1239
  }
unknown's avatar
unknown committed
1240

unknown's avatar
unknown committed
1241 1242
  if (res)
    DBUG_RETURN(res);
unknown's avatar
VIEW  
unknown committed
1243

unknown's avatar
unknown committed
1244 1245 1246
  if (!table)
    table= table_list->table;

1247
  if (!select_insert)
unknown's avatar
unknown committed
1248
  {
1249
    Item *fake_conds= 0;
1250
    TABLE_LIST *duplicate;
1251
    if ((duplicate= unique_table(thd, table_list, table_list->next_global, 1)))
1252
    {
1253
      update_non_unique_table_error(table_list, "INSERT", duplicate);
1254 1255
      DBUG_RETURN(TRUE);
    }
1256
    select_lex->fix_prepare_information(thd, &fake_conds, &fake_conds);
unknown's avatar
unknown committed
1257
    select_lex->first_execution= 0;
unknown's avatar
unknown committed
1258
  }
1259
  if (duplic == DUP_UPDATE || duplic == DUP_REPLACE)
1260
    table->prepare_for_position();
unknown's avatar
unknown committed
1261
  DBUG_RETURN(FALSE);
unknown's avatar
unknown committed
1262 1263 1264
}


unknown's avatar
unknown committed
1265 1266 1267 1268
	/* Check if there is more uniq keys after field */

static int last_uniq_key(TABLE *table,uint keynr)
{
1269
  while (++keynr < table->s->keys)
unknown's avatar
unknown committed
1270 1271 1272 1273 1274 1275 1276
    if (table->key_info[keynr].flags & HA_NOSAME)
      return 0;
  return 1;
}


/*
unknown's avatar
unknown committed
1277 1278 1279 1280 1281 1282 1283 1284 1285 1286
  Write a record to table with optional deleting of conflicting records,
  invoke proper triggers if needed.

  SYNOPSIS
     write_record()
      thd   - thread context
      table - table to which record should be written
      info  - COPY_INFO structure describing handling of duplicates
              and which is used for counting number of records inserted
              and deleted.
unknown's avatar
unknown committed
1287

unknown's avatar
unknown committed
1288 1289 1290 1291 1292
  NOTE
    Once this record will be written to table after insert trigger will
    be invoked. If instead of inserting new record we will update old one
    then both on update triggers will work instead. Similarly both on
    delete triggers will be invoked if we will delete conflicting records.
unknown's avatar
unknown committed
1293

1294
    Sets thd->no_trans_update.stmt to TRUE if table which is updated didn't have
unknown's avatar
unknown committed
1295 1296 1297 1298 1299
    transactions.

  RETURN VALUE
    0     - success
    non-0 - error
unknown's avatar
unknown committed
1300 1301 1302
*/


unknown's avatar
unknown committed
1303
int write_record(THD *thd, TABLE *table,COPY_INFO *info)
unknown's avatar
unknown committed
1304
{
unknown's avatar
unknown committed
1305
  int error, trg_error= 0;
unknown's avatar
unknown committed
1306
  char *key=0;
1307
  MY_BITMAP *save_read_set, *save_write_set;
1308 1309
  ulonglong prev_insert_id= table->file->next_insert_id;
  ulonglong insert_id_for_cur_row= 0;
1310
  DBUG_ENTER("write_record");
unknown's avatar
unknown committed
1311

unknown's avatar
unknown committed
1312
  info->records++;
1313 1314
  save_read_set=  table->read_set;
  save_write_set= table->write_set;
1315

1316 1317
  if (info->handle_duplicates == DUP_REPLACE ||
      info->handle_duplicates == DUP_UPDATE)
unknown's avatar
unknown committed
1318
  {
1319
    while ((error=table->file->ha_write_row(table->record[0])))
unknown's avatar
unknown committed
1320
    {
1321
      uint key_nr;
1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332
      /*
        If we do more than one iteration of this loop, from the second one the
        row will have an explicit value in the autoinc field, which was set at
        the first call of handler::update_auto_increment(). So we must save
        the autogenerated value to avoid thd->insert_id_for_cur_row to become
        0.
      */
      if (table->file->insert_id_for_cur_row > 0)
        insert_id_for_cur_row= table->file->insert_id_for_cur_row;
      else
        table->file->insert_id_for_cur_row= insert_id_for_cur_row;
1333
      bool is_duplicate_key_error;
1334
      if (table->file->is_fatal_error(error, HA_CHECK_DUP))
unknown's avatar
unknown committed
1335
	goto err;
1336
      is_duplicate_key_error= table->file->is_fatal_error(error, 0);
unknown's avatar
unknown committed
1337 1338
      if (!is_duplicate_key_error)
      {
1339 1340 1341 1342 1343
        /*
          We come here when we had an ignorable error which is not a duplicate
          key error. In this we ignore error if ignore flag is set, otherwise
          report error as usual. We will not do any duplicate key processing.
        */
unknown's avatar
unknown committed
1344
        if (info->ignore)
1345 1346
          goto ok_or_after_trg_err; /* Ignoring a not fatal error, return 0 */
        goto err;
unknown's avatar
unknown committed
1347
      }
unknown's avatar
unknown committed
1348 1349
      if ((int) (key_nr = table->file->get_dup_key(error)) < 0)
      {
1350
	error= HA_ERR_FOUND_DUPP_KEY;         /* Database can't find key */
unknown's avatar
unknown committed
1351 1352
	goto err;
      }
1353 1354
      /* Read all columns for the row we are going to replace */
      table->use_all_columns();
1355 1356 1357 1358 1359
      /*
	Don't allow REPLACE to replace a row when a auto_increment column
	was used.  This ensures that we don't get a problem when the
	whole range of the key has been used.
      */
1360 1361
      if (info->handle_duplicates == DUP_REPLACE &&
          table->next_number_field &&
1362
          key_nr == table->s->next_number_index &&
1363
	  (insert_id_for_cur_row > 0))
1364
	goto err;
1365
      if (table->file->ha_table_flags() & HA_DUPLICATE_POS)
unknown's avatar
unknown committed
1366
      {
1367
	if (table->file->rnd_pos(table->record[1],table->file->dup_ref))
unknown's avatar
unknown committed
1368 1369 1370 1371
	  goto err;
      }
      else
      {
1372
	if (table->file->extra(HA_EXTRA_FLUSH_CACHE)) /* Not needed with NISAM */
unknown's avatar
unknown committed
1373 1374 1375 1376
	{
	  error=my_errno;
	  goto err;
	}
unknown's avatar
unknown committed
1377

unknown's avatar
unknown committed
1378 1379
	if (!key)
	{
1380
	  if (!(key=(char*) my_safe_alloca(table->s->max_unique_length,
unknown's avatar
unknown committed
1381 1382 1383 1384 1385 1386
					   MAX_KEY_LENGTH)))
	  {
	    error=ENOMEM;
	    goto err;
	  }
	}
1387
	key_copy((uchar*) key,table->record[0],table->key_info+key_nr,0);
unknown's avatar
unknown committed
1388
	if ((error=(table->file->index_read_idx(table->record[1],key_nr,
1389
                                                (uchar*) key, HA_WHOLE_KEY,
unknown's avatar
unknown committed
1390 1391 1392
						HA_READ_KEY_EXACT))))
	  goto err;
      }
1393
      if (info->handle_duplicates == DUP_UPDATE)
unknown's avatar
unknown committed
1394
      {
unknown's avatar
unknown committed
1395
        int res= 0;
unknown's avatar
unknown committed
1396 1397 1398 1399
        /*
          We don't check for other UNIQUE keys - the first row
          that matches, is updated. If update causes a conflict again,
          an error is returned
1400
        */
unknown's avatar
unknown committed
1401
	DBUG_ASSERT(table->insert_values != NULL);
unknown's avatar
unknown committed
1402 1403
        store_record(table,insert_values);
        restore_record(table,record[1]);
unknown's avatar
unknown committed
1404 1405
        DBUG_ASSERT(info->update_fields->elements ==
                    info->update_values->elements);
unknown's avatar
unknown committed
1406
        if (fill_record_n_invoke_before_triggers(thd, *info->update_fields,
unknown's avatar
unknown committed
1407 1408
                                                 *info->update_values,
                                                 info->ignore,
unknown's avatar
unknown committed
1409 1410 1411
                                                 table->triggers,
                                                 TRG_EVENT_UPDATE))
          goto before_trg_err;
1412 1413

        /* CHECK OPTION for VIEW ... ON DUPLICATE KEY UPDATE ... */
unknown's avatar
unknown committed
1414 1415 1416
        if (info->view &&
            (res= info->view->view_check_option(current_thd, info->ignore)) ==
            VIEW_CHECK_SKIP)
unknown's avatar
unknown committed
1417
          goto ok_or_after_trg_err;
unknown's avatar
unknown committed
1418
        if (res == VIEW_CHECK_ERROR)
unknown's avatar
unknown committed
1419
          goto before_trg_err;
1420

1421
        table->file->restore_auto_increment(prev_insert_id);
1422 1423
        if ((table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ &&
             !bitmap_is_subset(table->write_set, table->read_set)) ||
unknown's avatar
unknown committed
1424
            compare_record(table))
1425
        {
1426
          if ((error=table->file->ha_update_row(table->record[1],
1427 1428
                                                table->record[0])) &&
              error != HA_ERR_RECORD_IS_THE_SAME)
1429
          {
1430 1431
            if (info->ignore &&
                !table->file->is_fatal_error(error, HA_CHECK_DUP_KEY))
1432 1433 1434 1435
            {
              goto ok_or_after_trg_err;
            }
            goto err;
1436
          }
unknown's avatar
unknown committed
1437

1438 1439 1440 1441
          if (error != HA_ERR_RECORD_IS_THE_SAME)
            info->updated++;
          else
            error= 0;
unknown's avatar
unknown committed
1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452
          /*
            If ON DUP KEY UPDATE updates a row instead of inserting one, it's
            like a regular UPDATE statement: it should not affect the value of a
            next SELECT LAST_INSERT_ID() or mysql_insert_id().
            Except if LAST_INSERT_ID(#) was in the INSERT query, which is
            handled separately by THD::arg_of_last_insert_id_function.
          */
          insert_id_for_cur_row= table->file->insert_id_for_cur_row= 0;
          if (table->next_number_field)
            table->file->adjust_next_insert_id_after_explicit_value(
              table->next_number_field->val_int());
unknown's avatar
unknown committed
1453 1454
          trg_error= (table->triggers &&
                      table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
unknown's avatar
unknown committed
1455
                                                        TRG_ACTION_AFTER, TRUE));
1456 1457
          info->copied++;
        }
1458

1459 1460 1461 1462 1463
        if (table->next_number_field)
          table->file->adjust_next_insert_id_after_explicit_value(
            table->next_number_field->val_int());
        info->touched++;

unknown's avatar
unknown committed
1464
        goto ok_or_after_trg_err;
1465 1466 1467
      }
      else /* DUP_REPLACE */
      {
unknown's avatar
unknown committed
1468 1469 1470 1471 1472
	/*
	  The manual defines the REPLACE semantics that it is either
	  an INSERT or DELETE(s) + INSERT; FOREIGN KEY checks in
	  InnoDB do not function in the defined way if we allow MySQL
	  to convert the latter operation internally to an UPDATE.
1473 1474
          We also should not perform this conversion if we have 
          timestamp field with ON UPDATE which is different from DEFAULT.
1475 1476 1477 1478 1479 1480
          Another case when conversion should not be performed is when
          we have ON DELETE trigger on table so user may notice that
          we cheat here. Note that it is ok to do such conversion for
          tables which have ON UPDATE but have no ON DELETE triggers,
          we just should not expose this fact to users by invoking
          ON UPDATE triggers.
unknown's avatar
unknown committed
1481 1482
	*/
	if (last_uniq_key(table,key_nr) &&
1483
	    !table->file->referenced_by_foreign_key() &&
1484
            (table->timestamp_field_type == TIMESTAMP_NO_AUTO_SET ||
1485 1486
             table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH) &&
            (!table->triggers || !table->triggers->has_delete_triggers()))
1487
        {
1488
          if ((error=table->file->ha_update_row(table->record[1],
1489 1490
					        table->record[0])) &&
              error != HA_ERR_RECORD_IS_THE_SAME)
1491
            goto err;
1492 1493 1494 1495
          if (error != HA_ERR_RECORD_IS_THE_SAME)
            info->deleted++;
          else
            error= 0;
1496
          thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
1497 1498 1499 1500 1501
          /*
            Since we pretend that we have done insert we should call
            its after triggers.
          */
          goto after_trg_n_copied_inc;
unknown's avatar
unknown committed
1502 1503 1504 1505 1506 1507 1508
        }
        else
        {
          if (table->triggers &&
              table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                                TRG_ACTION_BEFORE, TRUE))
            goto before_trg_err;
1509
          if ((error=table->file->ha_delete_row(table->record[1])))
unknown's avatar
unknown committed
1510 1511 1512
            goto err;
          info->deleted++;
          if (!table->file->has_transactions())
unknown's avatar
unknown committed
1513
            thd->no_trans_update.stmt= TRUE;
unknown's avatar
unknown committed
1514 1515 1516 1517 1518 1519 1520 1521
          if (table->triggers &&
              table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                                TRG_ACTION_AFTER, TRUE))
          {
            trg_error= 1;
            goto ok_or_after_trg_err;
          }
          /* Let us attempt do write_row() once more */
1522
        }
unknown's avatar
unknown committed
1523 1524
      }
    }
1525
    thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
1526 1527 1528 1529 1530 1531 1532
    /*
      Restore column maps if they where replaced during an duplicate key
      problem.
    */
    if (table->read_set != save_read_set ||
        table->write_set != save_write_set)
      table->column_bitmaps_set(save_read_set, save_write_set);
unknown's avatar
unknown committed
1533
  }
1534
  else if ((error=table->file->ha_write_row(table->record[0])))
unknown's avatar
unknown committed
1535
  {
1536
    if (!info->ignore ||
1537
        table->file->is_fatal_error(error, HA_CHECK_DUP))
unknown's avatar
unknown committed
1538
      goto err;
1539
    table->file->restore_auto_increment(prev_insert_id);
1540
    goto ok_or_after_trg_err;
unknown's avatar
unknown committed
1541
  }
1542 1543 1544

after_trg_n_copied_inc:
  info->copied++;
1545
  thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
1546 1547 1548
  trg_error= (table->triggers &&
              table->triggers->process_triggers(thd, TRG_EVENT_INSERT,
                                                TRG_ACTION_AFTER, TRUE));
unknown's avatar
unknown committed
1549 1550

ok_or_after_trg_err:
unknown's avatar
unknown committed
1551
  if (key)
1552
    my_safe_afree(key,table->s->max_unique_length,MAX_KEY_LENGTH);
unknown's avatar
unknown committed
1553
  if (!table->file->has_transactions())
1554
    thd->no_trans_update.stmt= TRUE;
unknown's avatar
unknown committed
1555
  DBUG_RETURN(trg_error);
unknown's avatar
unknown committed
1556 1557

err:
1558
  info->last_errno= error;
1559 1560 1561
  /* current_select is NULL if this is a delayed insert */
  if (thd->lex->current_select)
    thd->lex->current_select->no_error= 0;        // Give error
unknown's avatar
unknown committed
1562
  table->file->print_error(error,MYF(0));
1563
  
unknown's avatar
unknown committed
1564
before_trg_err:
1565
  table->file->restore_auto_increment(prev_insert_id);
unknown's avatar
unknown committed
1566 1567
  if (key)
    my_safe_afree(key, table->s->max_unique_length, MAX_KEY_LENGTH);
1568
  table->column_bitmaps_set(save_read_set, save_write_set);
1569
  DBUG_RETURN(1);
unknown's avatar
unknown committed
1570 1571 1572 1573
}


/******************************************************************************
unknown's avatar
unknown committed
1574
  Check that all fields with arn't null_fields are used
unknown's avatar
unknown committed
1575 1576
******************************************************************************/

1577 1578
int check_that_all_fields_are_given_values(THD *thd, TABLE *entry,
                                           TABLE_LIST *table_list)
unknown's avatar
unknown committed
1579
{
1580
  int err= 0;
1581 1582
  MY_BITMAP *write_set= entry->write_set;

unknown's avatar
unknown committed
1583 1584
  for (Field **field=entry->field ; *field ; field++)
  {
1585
    if (!bitmap_is_set(write_set, (*field)->field_index) &&
1586
        ((*field)->flags & NO_DEFAULT_VALUE_FLAG) &&
1587
        ((*field)->real_type() != MYSQL_TYPE_ENUM))
unknown's avatar
unknown committed
1588
    {
1589 1590 1591
      bool view= FALSE;
      if (table_list)
      {
unknown's avatar
unknown committed
1592
        table_list= table_list->top_table();
unknown's avatar
unknown committed
1593
        view= test(table_list->view);
1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609
      }
      if (view)
      {
        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                            ER_NO_DEFAULT_FOR_VIEW_FIELD,
                            ER(ER_NO_DEFAULT_FOR_VIEW_FIELD),
                            table_list->view_db.str,
                            table_list->view_name.str);
      }
      else
      {
        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                            ER_NO_DEFAULT_FOR_FIELD,
                            ER(ER_NO_DEFAULT_FOR_FIELD),
                            (*field)->field_name);
      }
1610
      err= 1;
unknown's avatar
unknown committed
1611 1612
    }
  }
1613
  return thd->abort_on_warning ? err : 0;
unknown's avatar
unknown committed
1614 1615 1616
}

/*****************************************************************************
unknown's avatar
unknown committed
1617 1618
  Handling of delayed inserts
  A thread is created for each table that one uses with the DELAYED attribute.
unknown's avatar
unknown committed
1619 1620
*****************************************************************************/

1621 1622
#ifndef EMBEDDED_LIBRARY

unknown's avatar
unknown committed
1623 1624
class delayed_row :public ilink {
public:
1625
  char *record;
unknown's avatar
unknown committed
1626 1627
  enum_duplicates dup;
  time_t start_time;
1628 1629 1630
  bool query_start_used, ignore, log_query;
  bool stmt_depends_on_first_successful_insert_id_in_prev_stmt;
  ulonglong first_successful_insert_id_in_prev_stmt;
1631
  ulonglong forced_insert_id;
1632 1633
  ulong auto_increment_increment;
  ulong auto_increment_offset;
1634
  timestamp_auto_set_type timestamp_field_type;
1635
  LEX_STRING query;
unknown's avatar
unknown committed
1636

1637 1638 1639
  delayed_row(LEX_STRING const query_arg, enum_duplicates dup_arg,
              bool ignore_arg, bool log_query_arg)
    : record(0), dup(dup_arg), ignore(ignore_arg), log_query(log_query_arg),
1640
      forced_insert_id(0), query(query_arg)
1641
    {}
unknown's avatar
unknown committed
1642 1643
  ~delayed_row()
  {
1644
    x_free(query.str);
unknown's avatar
unknown committed
1645 1646 1647 1648
    x_free(record);
  }
};

1649
/**
1650
  Delayed_insert - context of a thread responsible for delayed insert
1651 1652 1653 1654
  into one table. When processing delayed inserts, we create an own
  thread for every distinct table. Later on all delayed inserts directed
  into that table are handled by a dedicated thread.
*/
unknown's avatar
unknown committed
1655

1656
class Delayed_insert :public ilink {
unknown's avatar
unknown committed
1657 1658 1659 1660 1661 1662 1663 1664 1665 1666
  uint locks_in_memory;
public:
  THD thd;
  TABLE *table;
  pthread_mutex_t mutex;
  pthread_cond_t cond,cond_client;
  volatile uint tables_in_use,stacked_inserts;
  volatile bool status,dead;
  COPY_INFO info;
  I_List<delayed_row> rows;
1667
  ulong group_count;
unknown's avatar
unknown committed
1668
  TABLE_LIST table_list;			// Argument
unknown's avatar
unknown committed
1669

1670
  Delayed_insert()
1671
    :locks_in_memory(0),
unknown's avatar
unknown committed
1672 1673 1674
     table(0),tables_in_use(0),stacked_inserts(0), status(0), dead(0),
     group_count(0)
  {
1675 1676
    thd.security_ctx->user=thd.security_ctx->priv_user=(char*) delayed_user;
    thd.security_ctx->host=(char*) my_localhost;
unknown's avatar
unknown committed
1677 1678 1679
    thd.current_tablenr=0;
    thd.version=refresh_version;
    thd.command=COM_DELAYED_INSERT;
1680 1681
    thd.lex->current_select= 0; 		// for my_message_sql
    thd.lex->sql_command= SQLCOM_INSERT;        // For innodb::store_lock()
1682 1683 1684 1685
    /*
      Statement-based replication of INSERT DELAYED has problems with RAND()
      and user vars, so in mixed mode we go to row-based.
    */
1686
    thd.lex->set_stmt_unsafe();
1687
    thd.set_current_stmt_binlog_row_based_if_mixed();
unknown's avatar
unknown committed
1688

1689 1690
    bzero((char*) &thd.net, sizeof(thd.net));		// Safety
    bzero((char*) &table_list, sizeof(table_list));	// Safety
1691
    thd.system_thread= SYSTEM_THREAD_DELAYED_INSERT;
1692
    thd.security_ctx->host_or_ip= "";
unknown's avatar
unknown committed
1693
    bzero((char*) &info,sizeof(info));
1694
    pthread_mutex_init(&mutex,MY_MUTEX_INIT_FAST);
unknown's avatar
unknown committed
1695 1696 1697 1698 1699 1700
    pthread_cond_init(&cond,NULL);
    pthread_cond_init(&cond_client,NULL);
    VOID(pthread_mutex_lock(&LOCK_thread_count));
    delayed_insert_threads++;
    VOID(pthread_mutex_unlock(&LOCK_thread_count));
  }
1701
  ~Delayed_insert()
unknown's avatar
unknown committed
1702
  {
1703
    /* The following is not really needed, but just for safety */
unknown's avatar
unknown committed
1704 1705 1706 1707 1708 1709
    delayed_row *row;
    while ((row=rows.get()))
      delete row;
    if (table)
      close_thread_tables(&thd);
    VOID(pthread_mutex_lock(&LOCK_thread_count));
unknown's avatar
unknown committed
1710 1711 1712
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    pthread_cond_destroy(&cond_client);
unknown's avatar
unknown committed
1713 1714
    thd.unlink();				// Must be unlinked under lock
    x_free(thd.query);
1715
    thd.security_ctx->user= thd.security_ctx->host=0;
unknown's avatar
unknown committed
1716 1717 1718
    thread_count--;
    delayed_insert_threads--;
    VOID(pthread_mutex_unlock(&LOCK_thread_count));
1719
    VOID(pthread_cond_broadcast(&COND_thread_count)); /* Tell main we are ready */
unknown's avatar
unknown committed
1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748
  }

  /* The following is for checking when we can delete ourselves */
  inline void lock()
  {
    locks_in_memory++;				// Assume LOCK_delay_insert
  }
  void unlock()
  {
    pthread_mutex_lock(&LOCK_delayed_insert);
    if (!--locks_in_memory)
    {
      pthread_mutex_lock(&mutex);
      if (thd.killed && ! stacked_inserts && ! tables_in_use)
      {
	pthread_cond_signal(&cond);
	status=1;
      }
      pthread_mutex_unlock(&mutex);
    }
    pthread_mutex_unlock(&LOCK_delayed_insert);
  }
  inline uint lock_count() { return locks_in_memory; }

  TABLE* get_local_table(THD* client_thd);
  bool handle_inserts(void);
};


1749
I_List<Delayed_insert> delayed_threads;
unknown's avatar
unknown committed
1750 1751


1752 1753 1754 1755
/**
  Return an instance of delayed insert thread that can handle
  inserts into a given table, if it exists. Otherwise return NULL.
*/
unknown's avatar
unknown committed
1756

1757
static
1758
Delayed_insert *find_handler(THD *thd, TABLE_LIST *table_list)
unknown's avatar
unknown committed
1759 1760 1761
{
  thd->proc_info="waiting for delay_list";
  pthread_mutex_lock(&LOCK_delayed_insert);	// Protect master list
1762 1763
  I_List_iterator<Delayed_insert> it(delayed_threads);
  Delayed_insert *tmp;
unknown's avatar
unknown committed
1764 1765
  while ((tmp=it++))
  {
unknown's avatar
unknown committed
1766 1767
    if (!strcmp(tmp->thd.db, table_list->db) &&
	!strcmp(table_list->table_name, tmp->table->s->table_name.str))
unknown's avatar
unknown committed
1768 1769 1770 1771 1772 1773 1774 1775 1776 1777
    {
      tmp->lock();
      break;
    }
  }
  pthread_mutex_unlock(&LOCK_delayed_insert); // For unlink from list
  return tmp;
}


1778 1779 1780 1781
/**
  Attempt to find or create a delayed insert thread to handle inserts
  into this table.

unknown's avatar
unknown committed
1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800
  @return In case of success, table_list->table points to a local copy
          of the delayed table or is set to NULL, which indicates a
          request for lock upgrade. In case of failure, value of
          table_list->table is undefined.
  @retval TRUE  - this thread ran out of resources OR
                - a newly created delayed insert thread ran out of
                  resources OR
                - the created thread failed to open and lock the table
                  (e.g. because it does not exist) OR
                - the table opened in the created thread turned out to
                  be a view
  @retval FALSE - table successfully opened OR
                - too many delayed insert threads OR
                - the table has triggers and we have to fall back to
                  a normal INSERT
                Two latter cases indicate a request for lock upgrade.

  XXX: why do we regard INSERT DELAYED into a view as an error and
  do not simply a lock upgrade?
1801 1802
*/

unknown's avatar
unknown committed
1803 1804
static
bool delayed_get_table(THD *thd, TABLE_LIST *table_list)
unknown's avatar
unknown committed
1805 1806
{
  int error;
1807
  Delayed_insert *tmp;
unknown's avatar
unknown committed
1808 1809
  DBUG_ENTER("delayed_get_table");

unknown's avatar
unknown committed
1810 1811
  /* Must be set in the parser */
  DBUG_ASSERT(table_list->db);
unknown's avatar
unknown committed
1812

1813
  /* Find the thread which handles this table. */
unknown's avatar
unknown committed
1814 1815
  if (!(tmp=find_handler(thd,table_list)))
  {
1816 1817 1818 1819
    /*
      No match. Create a new thread to handle the table, but
      no more than max_insert_delayed_threads.
    */
1820
    if (delayed_insert_threads >= thd->variables.max_insert_delayed_threads)
1821
      DBUG_RETURN(0);
unknown's avatar
unknown committed
1822 1823
    thd->proc_info="Creating delayed handler";
    pthread_mutex_lock(&LOCK_delayed_create);
1824 1825 1826 1827 1828
    /*
      The first search above was done without LOCK_delayed_create.
      Another thread might have created the handler in between. Search again.
    */
    if (! (tmp= find_handler(thd, table_list)))
unknown's avatar
unknown committed
1829
    {
1830
      if (!(tmp=new Delayed_insert()))
unknown's avatar
unknown committed
1831
      {
1832
	my_error(ER_OUTOFMEMORY,MYF(0),sizeof(Delayed_insert));
unknown's avatar
unknown committed
1833 1834
        thd->fatal_error();
        goto end_create;
unknown's avatar
unknown committed
1835
      }
unknown's avatar
unknown committed
1836 1837 1838
      pthread_mutex_lock(&LOCK_thread_count);
      thread_count++;
      pthread_mutex_unlock(&LOCK_thread_count);
unknown's avatar
unknown committed
1839 1840 1841
      tmp->thd.set_db(table_list->db, strlen(table_list->db));
      tmp->thd.query= my_strdup(table_list->table_name,MYF(MY_WME));
      if (tmp->thd.db == NULL || tmp->thd.query == NULL)
unknown's avatar
unknown committed
1842
      {
unknown's avatar
unknown committed
1843
        /* The error is reported */
unknown's avatar
unknown committed
1844
	delete tmp;
unknown's avatar
unknown committed
1845 1846
        thd->fatal_error();
        goto end_create;
unknown's avatar
unknown committed
1847
      }
unknown's avatar
unknown committed
1848
      tmp->table_list= *table_list;			// Needed to open table
1849
      tmp->table_list.alias= tmp->table_list.table_name= tmp->thd.query;
unknown's avatar
unknown committed
1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860
      tmp->lock();
      pthread_mutex_lock(&tmp->mutex);
      if ((error=pthread_create(&tmp->thd.real_id,&connection_attrib,
				handle_delayed_insert,(void*) tmp)))
      {
	DBUG_PRINT("error",
		   ("Can't create thread to handle delayed insert (error %d)",
		    error));
	pthread_mutex_unlock(&tmp->mutex);
	tmp->unlock();
	delete tmp;
unknown's avatar
unknown committed
1861
	my_error(ER_CANT_CREATE_THREAD, MYF(0), error);
unknown's avatar
unknown committed
1862 1863
        thd->fatal_error();
        goto end_create;
unknown's avatar
unknown committed
1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875
      }

      /* Wait until table is open */
      thd->proc_info="waiting for handler open";
      while (!tmp->thd.killed && !tmp->table && !thd->killed)
      {
	pthread_cond_wait(&tmp->cond_client,&tmp->mutex);
      }
      pthread_mutex_unlock(&tmp->mutex);
      thd->proc_info="got old table";
      if (tmp->thd.killed)
      {
1876 1877
        if (tmp->thd.net.report_error)
        {
unknown's avatar
unknown committed
1878 1879 1880 1881 1882 1883 1884 1885
          /*
            Copy the error message. Note that we don't treat fatal
            errors in the delayed thread as fatal errors in the
            main thread. Use of my_message will enable stored
            procedures continue handlers.
          */
          my_message(tmp->thd.net.last_errno, tmp->thd.net.last_error,
                     MYF(0));
unknown's avatar
unknown committed
1886 1887
	}
	tmp->unlock();
unknown's avatar
unknown committed
1888
        goto end_create;
unknown's avatar
unknown committed
1889 1890 1891 1892
      }
      if (thd->killed)
      {
	tmp->unlock();
1893
	goto end_create;
unknown's avatar
unknown committed
1894 1895 1896 1897 1898 1899
      }
    }
    pthread_mutex_unlock(&LOCK_delayed_create);
  }

  pthread_mutex_lock(&tmp->mutex);
unknown's avatar
unknown committed
1900
  table_list->table= tmp->get_local_table(thd);
unknown's avatar
unknown committed
1901
  pthread_mutex_unlock(&tmp->mutex);
unknown's avatar
unknown committed
1902 1903
  if (table_list->table)
  {
unknown's avatar
unknown committed
1904
    DBUG_ASSERT(thd->net.report_error == 0);
unknown's avatar
unknown committed
1905
    thd->di=tmp;
unknown's avatar
unknown committed
1906
  }
1907 1908
  /* Unlock the delayed insert object after its last access. */
  tmp->unlock();
1909
  DBUG_RETURN((table_list->table == NULL));
1910

unknown's avatar
unknown committed
1911
end_create:
1912
  pthread_mutex_unlock(&LOCK_delayed_create);
unknown's avatar
unknown committed
1913
  DBUG_RETURN(thd->net.report_error);
unknown's avatar
unknown committed
1914 1915 1916
}


1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927
/**
  As we can't let many client threads modify the same TABLE
  structure of the dedicated delayed insert thread, we create an
  own structure for each client thread. This includes a row
  buffer to save the column values and new fields that point to
  the new row buffer. The memory is allocated in the client
  thread and is freed automatically.

  @pre This function is called from the client thread.  Delayed
       insert thread mutex must be acquired before invoking this
       function.
unknown's avatar
unknown committed
1928 1929 1930

  @return Not-NULL table object on success. NULL in case of an error,
                    which is set in client_thd.
unknown's avatar
unknown committed
1931 1932
*/

1933
TABLE *Delayed_insert::get_local_table(THD* client_thd)
unknown's avatar
unknown committed
1934 1935
{
  my_ptrdiff_t adjust_ptrs;
1936
  Field **field,**org_field, *found_next_number_field;
unknown's avatar
unknown committed
1937
  TABLE *copy;
unknown's avatar
unknown committed
1938
  TABLE_SHARE *share= table->s;
1939
  uchar *bitmap;
1940
  DBUG_ENTER("Delayed_insert::get_local_table");
unknown's avatar
unknown committed
1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957

  /* First request insert thread to get a lock */
  status=1;
  tables_in_use++;
  if (!thd.lock)				// Table is not locked
  {
    client_thd->proc_info="waiting for handler lock";
    pthread_cond_signal(&cond);			// Tell handler to lock table
    while (!dead && !thd.lock && ! client_thd->killed)
    {
      pthread_cond_wait(&cond_client,&mutex);
    }
    client_thd->proc_info="got handler lock";
    if (client_thd->killed)
      goto error;
    if (dead)
    {
unknown's avatar
unknown committed
1958
      my_message(thd.net.last_errno, thd.net.last_error, MYF(0));
unknown's avatar
unknown committed
1959 1960 1961 1962
      goto error;
    }
  }

1963 1964 1965 1966 1967 1968 1969
  /*
    Allocate memory for the TABLE object, the field pointers array, and
    one record buffer of reclength size. Normally a table has three
    record buffers of rec_buff_length size, which includes alignment
    bytes. Since the table copy is used for creating one record only,
    the other record buffers and alignment are unnecessary.
  */
unknown's avatar
unknown committed
1970
  client_thd->proc_info="allocating local table";
unknown's avatar
unknown committed
1971
  copy= (TABLE*) client_thd->alloc(sizeof(*copy)+
unknown's avatar
unknown committed
1972
				   (share->fields+1)*sizeof(Field**)+
1973 1974
				   share->reclength +
                                   share->column_bitmap_size*2);
unknown's avatar
unknown committed
1975 1976 1977
  if (!copy)
    goto error;

1978
  /* Copy the TABLE object. */
unknown's avatar
unknown committed
1979
  *copy= *table;
unknown's avatar
unknown committed
1980
  /* We don't need to change the file handler here */
1981 1982
  /* Assign the pointers for the field pointers array and the record. */
  field= copy->field= (Field**) (copy + 1);
1983
  bitmap= (uchar*) (field + share->fields + 1);
1984 1985
  copy->record[0]= (bitmap + share->column_bitmap_size * 2);
  memcpy((char*) copy->record[0], (char*) table->record[0], share->reclength);
1986 1987 1988 1989 1990 1991 1992 1993 1994 1995
  /*
    Make a copy of all fields.
    The copied fields need to point into the copied record. This is done
    by copying the field objects with their old pointer values and then
    "move" the pointers by the distance between the original and copied
    records. That way we preserve the relative positions in the records.
  */
  adjust_ptrs= PTR_BYTE_DIFF(copy->record[0], table->record[0]);
  found_next_number_field= table->found_next_number_field;
  for (org_field= table->field; *org_field; org_field++, field++)
unknown's avatar
unknown committed
1996
  {
1997
    if (!(*field= (*org_field)->new_field(client_thd->mem_root, copy, 1)))
unknown's avatar
unknown committed
1998
      goto error;
1999
    (*field)->orig_table= copy;			// Remove connection
unknown's avatar
unknown committed
2000
    (*field)->move_field_offset(adjust_ptrs);	// Point at copy->record[0]
2001 2002
    if (*org_field == found_next_number_field)
      (*field)->table->found_next_number_field= *field;
unknown's avatar
unknown committed
2003 2004 2005 2006 2007 2008 2009 2010
  }
  *field=0;

  /* Adjust timestamp */
  if (table->timestamp_field)
  {
    /* Restore offset as this may have been reset in handle_inserts */
    copy->timestamp_field=
unknown's avatar
unknown committed
2011
      (Field_timestamp*) copy->field[share->timestamp_field_offset];
2012
    copy->timestamp_field->unireg_check= table->timestamp_field->unireg_check;
2013
    copy->timestamp_field_type= copy->timestamp_field->get_auto_set_type();
unknown's avatar
unknown committed
2014 2015
  }

2016 2017
  /* Adjust in_use for pointing to client thread */
  copy->in_use= client_thd;
2018 2019 2020 2021

  /* Adjust lock_count. This table object is not part of a lock. */
  copy->lock_count= 0;

2022 2023 2024 2025 2026 2027 2028 2029 2030
  /* Adjust bitmaps */
  copy->def_read_set.bitmap= (my_bitmap_map*) bitmap;
  copy->def_write_set.bitmap= ((my_bitmap_map*)
                               (bitmap + share->column_bitmap_size));
  copy->tmp_set.bitmap= 0;                      // To catch errors
  bzero((char*) bitmap, share->column_bitmap_size*2);
  copy->read_set=  &copy->def_read_set;
  copy->write_set= &copy->def_write_set;

2031
  DBUG_RETURN(copy);
unknown's avatar
unknown committed
2032 2033 2034 2035 2036 2037

  /* Got fatal error */
 error:
  tables_in_use--;
  status=1;
  pthread_cond_signal(&cond);			// Inform thread about abort
2038
  DBUG_RETURN(0);
unknown's avatar
unknown committed
2039 2040 2041 2042 2043
}


/* Put a question in queue */

unknown's avatar
unknown committed
2044
static
unknown's avatar
unknown committed
2045 2046
int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
                  LEX_STRING query, bool ignore, bool log_on)
unknown's avatar
unknown committed
2047
{
2048
  delayed_row *row= 0;
2049
  Delayed_insert *di=thd->di;
2050
  const Discrete_interval *forced_auto_inc;
unknown's avatar
unknown committed
2051
  DBUG_ENTER("write_delayed");
2052 2053
  DBUG_PRINT("enter", ("query = '%s' length %lu", query.str,
                       (ulong) query.length));
unknown's avatar
unknown committed
2054 2055 2056 2057 2058 2059 2060

  thd->proc_info="waiting for handler insert";
  pthread_mutex_lock(&di->mutex);
  while (di->stacked_inserts >= delayed_queue_size && !thd->killed)
    pthread_cond_wait(&di->cond_client,&di->mutex);
  thd->proc_info="storing row into queue";

2061
  if (thd->killed)
unknown's avatar
unknown committed
2062 2063
    goto err;

2064 2065 2066 2067 2068 2069 2070
  /*
    Take a copy of the query string, if there is any. The string will
    be free'ed when the row is destroyed. If there is no query string,
    we don't do anything special.
   */

  if (query.str)
2071 2072 2073
  {
    char *str;
    if (!(str= my_strndup(query.str, query.length, MYF(MY_WME))))
2074
      goto err;
2075 2076
    query.str= str;
  }
2077 2078 2079 2080
  row= new delayed_row(query, duplic, ignore, log_on);
  if (row == NULL)
  {
    my_free(query.str, MYF(MY_WME));
unknown's avatar
unknown committed
2081
    goto err;
2082
  }
unknown's avatar
unknown committed
2083

2084
  if (!(row->record= (char*) my_malloc(table->s->reclength, MYF(MY_WME))))
unknown's avatar
unknown committed
2085
    goto err;
2086
  memcpy(row->record, table->record[0], table->s->reclength);
unknown's avatar
unknown committed
2087 2088
  row->start_time=		thd->start_time;
  row->query_start_used=	thd->query_start_used;
2089 2090 2091 2092 2093 2094 2095 2096 2097 2098
  /*
    those are for the binlog: LAST_INSERT_ID() has been evaluated at this
    time, so record does not need it, but statement-based binlogging of the
    INSERT will need when the row is actually inserted.
    As for SET INSERT_ID, DELAYED does not honour it (BUG#20830).
  */
  row->stmt_depends_on_first_successful_insert_id_in_prev_stmt=
    thd->stmt_depends_on_first_successful_insert_id_in_prev_stmt;
  row->first_successful_insert_id_in_prev_stmt=
    thd->first_successful_insert_id_in_prev_stmt;
2099
  row->timestamp_field_type=    table->timestamp_field_type;
unknown's avatar
unknown committed
2100

2101
  /* Copy session variables. */
2102 2103
  row->auto_increment_increment= thd->variables.auto_increment_increment;
  row->auto_increment_offset=    thd->variables.auto_increment_offset;
2104 2105 2106 2107 2108 2109 2110
  /* Copy the next forced auto increment value, if any. */
  if ((forced_auto_inc= thd->auto_inc_intervals_forced.get_next()))
  {
    row->forced_insert_id= forced_auto_inc->minimum();
    DBUG_PRINT("delayed", ("transmitting auto_inc: %lu",
                           (ulong) row->forced_insert_id));
  }
2111

unknown's avatar
unknown committed
2112 2113 2114
  di->rows.push_back(row);
  di->stacked_inserts++;
  di->status=1;
2115
  if (table->s->blob_fields)
unknown's avatar
unknown committed
2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128
    unlink_blobs(table);
  pthread_cond_signal(&di->cond);

  thread_safe_increment(delayed_rows_in_use,&LOCK_delayed_status);
  pthread_mutex_unlock(&di->mutex);
  DBUG_RETURN(0);

 err:
  delete row;
  pthread_mutex_unlock(&di->mutex);
  DBUG_RETURN(1);
}

unknown's avatar
unknown committed
2129 2130 2131 2132
/**
  Signal the delayed insert thread that this user connection
  is finished using it for this statement.
*/
unknown's avatar
unknown committed
2133 2134 2135

static void end_delayed_insert(THD *thd)
{
2136
  DBUG_ENTER("end_delayed_insert");
2137
  Delayed_insert *di=thd->di;
unknown's avatar
unknown committed
2138
  pthread_mutex_lock(&di->mutex);
2139
  DBUG_PRINT("info",("tables in use: %d",di->tables_in_use));
unknown's avatar
unknown committed
2140 2141 2142 2143 2144 2145
  if (!--di->tables_in_use || di->thd.killed)
  {						// Unlock table
    di->status=1;
    pthread_cond_signal(&di->cond);
  }
  pthread_mutex_unlock(&di->mutex);
2146
  DBUG_VOID_RETURN;
unknown's avatar
unknown committed
2147 2148 2149 2150 2151 2152 2153 2154 2155
}


/* We kill all delayed threads when doing flush-tables */

void kill_delayed_threads(void)
{
  VOID(pthread_mutex_lock(&LOCK_delayed_insert)); // For unlink from list

2156 2157
  I_List_iterator<Delayed_insert> it(delayed_threads);
  Delayed_insert *tmp;
unknown's avatar
unknown committed
2158 2159
  while ((tmp=it++))
  {
unknown's avatar
SCRUM  
unknown committed
2160
    tmp->thd.killed= THD::KILL_CONNECTION;
unknown's avatar
unknown committed
2161 2162 2163
    if (tmp->thd.mysys_var)
    {
      pthread_mutex_lock(&tmp->thd.mysys_var->mutex);
unknown's avatar
unknown committed
2164
      if (tmp->thd.mysys_var->current_cond)
unknown's avatar
unknown committed
2165
      {
unknown's avatar
unknown committed
2166 2167 2168 2169 2170 2171
	/*
	  We need the following test because the main mutex may be locked
	  in handle_delayed_insert()
	*/
	if (&tmp->mutex != tmp->thd.mysys_var->current_mutex)
	  pthread_mutex_lock(tmp->thd.mysys_var->current_mutex);
unknown's avatar
unknown committed
2172
	pthread_cond_broadcast(tmp->thd.mysys_var->current_cond);
unknown's avatar
unknown committed
2173 2174
	if (&tmp->mutex != tmp->thd.mysys_var->current_mutex)
	  pthread_mutex_unlock(tmp->thd.mysys_var->current_mutex);
unknown's avatar
unknown committed
2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186
      }
      pthread_mutex_unlock(&tmp->thd.mysys_var->mutex);
    }
  }
  VOID(pthread_mutex_unlock(&LOCK_delayed_insert)); // For unlink from list
}


/*
 * Create a new delayed insert thread
*/

2187
pthread_handler_t handle_delayed_insert(void *arg)
unknown's avatar
unknown committed
2188
{
2189
  Delayed_insert *di=(Delayed_insert*) arg;
unknown's avatar
unknown committed
2190 2191 2192 2193 2194
  THD *thd= &di->thd;

  pthread_detach_this_thread();
  /* Add thread to THD list so that's it's visible in 'show processlist' */
  pthread_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
2195
  thd->thread_id= thd->variables.pseudo_thread_id= thread_id++;
2196
  thd->end_time();
unknown's avatar
unknown committed
2197
  threads.append(thd);
unknown's avatar
SCRUM  
unknown committed
2198
  thd->killed=abort_loop ? THD::KILL_CONNECTION : THD::NOT_KILLED;
unknown's avatar
unknown committed
2199 2200
  pthread_mutex_unlock(&LOCK_thread_count);

2201 2202 2203 2204 2205 2206 2207 2208
  /*
    Wait until the client runs into pthread_cond_wait(),
    where we free it after the table is opened and di linked in the list.
    If we did not wait here, the client might detect the opened table
    before it is linked to the list. It would release LOCK_delayed_create
    and allow another thread to create another handler for the same table,
    since it does not find one in the list.
  */
unknown's avatar
unknown committed
2209
  pthread_mutex_lock(&di->mutex);
2210
#if !defined( __WIN__) /* Win32 calls this in pthread_create */
unknown's avatar
unknown committed
2211 2212 2213 2214 2215 2216 2217 2218
  if (my_thread_init())
  {
    strmov(thd->net.last_error,ER(thd->net.last_errno=ER_OUT_OF_RESOURCES));
    goto end;
  }
#endif

  DBUG_ENTER("handle_delayed_insert");
2219
  thd->thread_stack= (char*) &thd;
unknown's avatar
unknown committed
2220
  if (init_thr_lock() || thd->store_globals())
unknown's avatar
unknown committed
2221
  {
2222
    thd->fatal_error();
unknown's avatar
unknown committed
2223
    strmov(thd->net.last_error,ER(thd->net.last_errno=ER_OUT_OF_RESOURCES));
2224
    goto err;
unknown's avatar
unknown committed
2225 2226 2227
  }

  /* open table */
unknown's avatar
unknown committed
2228
  if (!(di->table=open_ltable(thd,&di->table_list,TL_WRITE_DELAYED)))
unknown's avatar
unknown committed
2229
  {
2230
    thd->fatal_error();				// Abort waiting inserts
2231
    goto err;
unknown's avatar
unknown committed
2232
  }
2233
  if (!(di->table->file->ha_table_flags() & HA_CAN_INSERT_DELAYED))
unknown's avatar
unknown committed
2234
  {
2235
    thd->fatal_error();
2236
    my_error(ER_ILLEGAL_HA, MYF(0), di->table_list.table_name);
2237
    goto err;
unknown's avatar
unknown committed
2238
  }
unknown's avatar
unknown committed
2239 2240 2241 2242 2243 2244 2245
  if (di->table->triggers)
  {
    /*
      Table has triggers. This is not an error, but we do
      not support triggers with delayed insert. Terminate the delayed
      thread without an error and thus request lock upgrade.
    */
2246
    goto err;
unknown's avatar
unknown committed
2247
  }
unknown's avatar
unknown committed
2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262
  di->table->copy_blobs=1;

  /* One can now use this */
  pthread_mutex_lock(&LOCK_delayed_insert);
  delayed_threads.append(di);
  pthread_mutex_unlock(&LOCK_delayed_insert);

  /* Tell client that the thread is initialized */
  pthread_cond_signal(&di->cond_client);

  /* Now wait until we get an insert or lock to handle */
  /* We will not abort as long as a client thread uses this thread */

  for (;;)
  {
unknown's avatar
SCRUM  
unknown committed
2263
    if (thd->killed == THD::KILL_CONNECTION)
unknown's avatar
unknown committed
2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282
    {
      uint lock_count;
      /*
	Remove this from delay insert list so that no one can request a
	table from this
      */
      pthread_mutex_unlock(&di->mutex);
      pthread_mutex_lock(&LOCK_delayed_insert);
      di->unlink();
      lock_count=di->lock_count();
      pthread_mutex_unlock(&LOCK_delayed_insert);
      pthread_mutex_lock(&di->mutex);
      if (!lock_count && !di->tables_in_use && !di->stacked_inserts)
	break;					// Time to die
    }

    if (!di->status && !di->stacked_inserts)
    {
      struct timespec abstime;
2283
      set_timespec(abstime, delayed_insert_timeout);
unknown's avatar
unknown committed
2284 2285 2286 2287

      /* Information for pthread_kill */
      di->thd.mysys_var->current_mutex= &di->mutex;
      di->thd.mysys_var->current_cond= &di->cond;
2288
      di->thd.proc_info="Waiting for INSERT";
unknown's avatar
unknown committed
2289

2290
      DBUG_PRINT("info",("Waiting for someone to insert rows"));
unknown's avatar
unknown committed
2291
      while (!thd->killed)
unknown's avatar
unknown committed
2292 2293
      {
	int error;
2294
#if defined(HAVE_BROKEN_COND_TIMEDWAIT)
unknown's avatar
unknown committed
2295 2296 2297 2298
	error=pthread_cond_wait(&di->cond,&di->mutex);
#else
	error=pthread_cond_timedwait(&di->cond,&di->mutex,&abstime);
#ifdef EXTRA_DEBUG
2299
	if (error && error != EINTR && error != ETIMEDOUT)
unknown's avatar
unknown committed
2300 2301 2302 2303 2304 2305 2306 2307 2308
	{
	  fprintf(stderr, "Got error %d from pthread_cond_timedwait\n",error);
	  DBUG_PRINT("error",("Got error %d from pthread_cond_timedwait",
			      error));
	}
#endif
#endif
	if (thd->killed || di->status)
	  break;
unknown's avatar
unknown committed
2309
	if (error == ETIMEDOUT || error == ETIME)
unknown's avatar
unknown committed
2310
	{
unknown's avatar
SCRUM  
unknown committed
2311
	  thd->killed= THD::KILL_CONNECTION;
unknown's avatar
unknown committed
2312 2313 2314
	  break;
	}
      }
unknown's avatar
unknown committed
2315 2316
      /* We can't lock di->mutex and mysys_var->mutex at the same time */
      pthread_mutex_unlock(&di->mutex);
unknown's avatar
unknown committed
2317 2318 2319 2320
      pthread_mutex_lock(&di->thd.mysys_var->mutex);
      di->thd.mysys_var->current_mutex= 0;
      di->thd.mysys_var->current_cond= 0;
      pthread_mutex_unlock(&di->thd.mysys_var->mutex);
unknown's avatar
unknown committed
2321
      pthread_mutex_lock(&di->mutex);
unknown's avatar
unknown committed
2322
    }
2323
    di->thd.proc_info=0;
unknown's avatar
unknown committed
2324 2325 2326

    if (di->tables_in_use && ! thd->lock)
    {
2327
      bool not_used;
2328 2329 2330 2331 2332 2333 2334 2335 2336 2337
      /*
        Request for new delayed insert.
        Lock the table, but avoid to be blocked by a global read lock.
        If we got here while a global read lock exists, then one or more
        inserts started before the lock was requested. These are allowed
        to complete their work before the server returns control to the
        client which requested the global read lock. The delayed insert
        handler will close the table and finish when the outstanding
        inserts are done.
      */
2338
      if (! (thd->lock= mysql_lock_tables(thd, &di->table, 1,
2339 2340
                                          MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK,
                                          &not_used)))
unknown's avatar
unknown committed
2341
      {
2342 2343 2344
	/* Fatal error */
	di->dead= 1;
	thd->killed= THD::KILL_CONNECTION;
unknown's avatar
unknown committed
2345 2346 2347 2348 2349 2350 2351
      }
      pthread_cond_broadcast(&di->cond_client);
    }
    if (di->stacked_inserts)
    {
      if (di->handle_inserts())
      {
2352 2353 2354
	/* Some fatal error */
	di->dead= 1;
	thd->killed= THD::KILL_CONNECTION;
unknown's avatar
unknown committed
2355 2356 2357 2358 2359
      }
    }
    di->status=0;
    if (!di->stacked_inserts && !di->tables_in_use && thd->lock)
    {
unknown's avatar
unknown committed
2360 2361 2362 2363
      /*
        No one is doing a insert delayed
        Unlock table so that other threads can use it
      */
unknown's avatar
unknown committed
2364 2365 2366
      MYSQL_LOCK *lock=thd->lock;
      thd->lock=0;
      pthread_mutex_unlock(&di->mutex);
2367 2368 2369 2370
      /*
        We need to release next_insert_id before unlocking. This is
        enforced by handler::ha_external_lock().
      */
2371
      di->table->file->ha_release_auto_increment();
unknown's avatar
unknown committed
2372 2373 2374 2375 2376 2377 2378 2379
      mysql_unlock_tables(thd, lock);
      di->group_count=0;
      pthread_mutex_lock(&di->mutex);
    }
    if (di->tables_in_use)
      pthread_cond_broadcast(&di->cond_client); // If waiting clients
  }

2380 2381 2382 2383 2384 2385 2386
err:
  /*
    mysql_lock_tables() can potentially start a transaction and write
    a table map. In the event of an error, that transaction has to be
    rolled back.  We only need to roll back a potential statement
    transaction, since real transactions are rolled back in
    close_thread_tables().
2387 2388 2389 2390

    TODO: This is not true any more, table maps are generated on the
    first call to ha_*_row() instead. Remove code that are used to
    cover for the case outlined above.
2391 2392 2393
   */
  ha_rollback_stmt(thd);

2394
#ifndef __WIN__
unknown's avatar
unknown committed
2395
end:
2396
#endif
unknown's avatar
unknown committed
2397 2398 2399 2400 2401 2402 2403
  /*
    di should be unlinked from the thread handler list and have no active
    clients
  */

  close_thread_tables(thd);			// Free the table
  di->table=0;
2404 2405
  di->dead= 1;                                  // If error
  thd->killed= THD::KILL_CONNECTION;	        // If error
unknown's avatar
unknown committed
2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439
  pthread_cond_broadcast(&di->cond_client);	// Safety
  pthread_mutex_unlock(&di->mutex);

  pthread_mutex_lock(&LOCK_delayed_create);	// Because of delayed_get_table
  pthread_mutex_lock(&LOCK_delayed_insert);	
  delete di;
  pthread_mutex_unlock(&LOCK_delayed_insert);
  pthread_mutex_unlock(&LOCK_delayed_create);  

  my_thread_end();
  pthread_exit(0);
  DBUG_RETURN(0);
}


/* Remove pointers from temporary fields to allocated values */

static void unlink_blobs(register TABLE *table)
{
  for (Field **ptr=table->field ; *ptr ; ptr++)
  {
    if ((*ptr)->flags & BLOB_FLAG)
      ((Field_blob *) (*ptr))->clear_temporary();
  }
}

/* Free blobs stored in current row */

static void free_delayed_insert_blobs(register TABLE *table)
{
  for (Field **ptr=table->field ; *ptr ; ptr++)
  {
    if ((*ptr)->flags & BLOB_FLAG)
    {
2440
      uchar *str;
unknown's avatar
unknown committed
2441 2442 2443 2444 2445 2446 2447 2448
      ((Field_blob *) (*ptr))->get_ptr(&str);
      my_free(str,MYF(MY_ALLOW_ZERO_PTR));
      ((Field_blob *) (*ptr))->reset();
    }
  }
}


2449
bool Delayed_insert::handle_inserts(void)
unknown's avatar
unknown committed
2450 2451
{
  int error;
2452
  ulong max_rows;
2453 2454
  bool using_ignore= 0, using_opt_replace= 0,
       using_bin_log= mysql_bin_log.is_open();
2455
  delayed_row *row;
unknown's avatar
unknown committed
2456 2457 2458 2459 2460 2461
  DBUG_ENTER("handle_inserts");

  /* Allow client to insert new rows */
  pthread_mutex_unlock(&mutex);

  table->next_number_field=table->found_next_number_field;
2462
  table->use_all_columns();
unknown's avatar
unknown committed
2463 2464 2465 2466 2467

  thd.proc_info="upgrading lock";
  if (thr_upgrade_write_delay_lock(*thd.lock->locks))
  {
    /* This can only happen if thread is killed by shutdown */
unknown's avatar
unknown committed
2468
    sql_print_error(ER(ER_DELAYED_CANT_CHANGE_LOCK),table->s->table_name.str);
unknown's avatar
unknown committed
2469 2470 2471 2472
    goto err;
  }

  thd.proc_info="insert";
2473
  max_rows= delayed_insert_limit;
unknown's avatar
unknown committed
2474
  if (thd.killed || table->needs_reopen_or_name_lock())
unknown's avatar
unknown committed
2475
  {
unknown's avatar
SCRUM  
unknown committed
2476
    thd.killed= THD::KILL_CONNECTION;
2477
    max_rows= ULONG_MAX;                     // Do as much as possible
unknown's avatar
unknown committed
2478 2479
  }

2480 2481 2482 2483 2484 2485 2486
  /*
    We can't use row caching when using the binary log because if
    we get a crash, then binary log will contain rows that are not yet
    written to disk, which will cause problems in replication.
  */
  if (!using_bin_log)
    table->file->extra(HA_EXTRA_WRITE_CACHE);
unknown's avatar
unknown committed
2487
  pthread_mutex_lock(&mutex);
2488

unknown's avatar
unknown committed
2489 2490 2491 2492
  while ((row=rows.get()))
  {
    stacked_inserts--;
    pthread_mutex_unlock(&mutex);
2493
    memcpy(table->record[0],row->record,table->s->reclength);
unknown's avatar
unknown committed
2494 2495 2496

    thd.start_time=row->start_time;
    thd.query_start_used=row->query_start_used;
2497 2498 2499 2500 2501
    /*
      To get the exact auto_inc interval to store in the binlog we must not
      use values from the previous interval (of the previous rows).
    */
    bool log_query= (row->log_query && row->query.str != NULL);
2502 2503 2504
    DBUG_PRINT("delayed", ("query: '%s'  length: %lu", row->query.str ?
                           row->query.str : "[NULL]",
                           (ulong) row->query.length));
2505 2506
    if (log_query)
    {
2507 2508 2509 2510 2511 2512 2513 2514
      /*
        This is the first value of an INSERT statement.
        It is the right place to clear a forced insert_id.
        This is usually done after the last value of an INSERT statement,
        but we won't know this in the insert delayed thread. But before
        the first value is sufficiently equivalent to after the last
        value of the previous statement.
      */
2515 2516 2517
      table->file->ha_release_auto_increment();
      thd.auto_inc_intervals_in_cur_stmt_for_binlog.empty();
    }
2518 2519 2520 2521
    thd.first_successful_insert_id_in_prev_stmt= 
      row->first_successful_insert_id_in_prev_stmt;
    thd.stmt_depends_on_first_successful_insert_id_in_prev_stmt= 
      row->stmt_depends_on_first_successful_insert_id_in_prev_stmt;
2522
    table->timestamp_field_type= row->timestamp_field_type;
unknown's avatar
unknown committed
2523

2524
    /* Copy the session variables. */
2525 2526
    thd.variables.auto_increment_increment= row->auto_increment_increment;
    thd.variables.auto_increment_offset=    row->auto_increment_offset;
2527 2528 2529 2530 2531 2532 2533
    /* Copy a forced insert_id, if any. */
    if (row->forced_insert_id)
    {
      DBUG_PRINT("delayed", ("received auto_inc: %lu",
                             (ulong) row->forced_insert_id));
      thd.force_one_auto_inc_interval(row->forced_insert_id);
    }
2534

2535
    info.ignore= row->ignore;
unknown's avatar
unknown committed
2536
    info.handle_duplicates= row->dup;
2537
    if (info.ignore ||
unknown's avatar
unknown committed
2538
	info.handle_duplicates != DUP_ERROR)
unknown's avatar
unknown committed
2539 2540 2541 2542
    {
      table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
      using_ignore=1;
    }
2543 2544 2545 2546 2547 2548 2549
    if (info.handle_duplicates == DUP_REPLACE &&
        (!table->triggers ||
         !table->triggers->has_delete_triggers()))
    {
      table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
      using_opt_replace= 1;
    }
unknown's avatar
unknown committed
2550
    thd.clear_error(); // reset error for binlog
unknown's avatar
unknown committed
2551
    if (write_record(&thd, table, &info))
unknown's avatar
unknown committed
2552
    {
2553
      info.error_count++;				// Ignore errors
2554
      thread_safe_increment(delayed_insert_errors,&LOCK_delayed_status);
unknown's avatar
unknown committed
2555
      row->log_query = 0;
unknown's avatar
unknown committed
2556
    }
2557

unknown's avatar
unknown committed
2558 2559 2560 2561 2562
    if (using_ignore)
    {
      using_ignore=0;
      table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
    }
2563 2564 2565 2566 2567
    if (using_opt_replace)
    {
      using_opt_replace= 0;
      table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
    }
2568

2569
    if (log_query && mysql_bin_log.is_open())
unknown's avatar
unknown committed
2570 2571 2572 2573 2574 2575 2576 2577 2578
    {
      /*
        If the query has several rows to insert, only the first row will come
        here. In row-based binlogging, this means that the first row will be
        written to binlog as one Table_map event and one Rows event (due to an
        event flush done in binlog_query()), then all other rows of this query
        will be binlogged together as one single Table_map event and one
        single Rows event.
      */
2579 2580 2581
      thd.binlog_query(THD::ROW_QUERY_TYPE,
                       row->query.str, row->query.length,
                       FALSE, FALSE);
unknown's avatar
unknown committed
2582
    }
2583

2584
    if (table->s->blob_fields)
unknown's avatar
unknown committed
2585
      free_delayed_insert_blobs(table);
2586
    thread_safe_decrement(delayed_rows_in_use,&LOCK_delayed_status);
unknown's avatar
unknown committed
2587 2588 2589 2590
    thread_safe_increment(delayed_insert_writes,&LOCK_delayed_status);
    pthread_mutex_lock(&mutex);

    delete row;
2591 2592 2593 2594 2595 2596 2597
    /*
      Let READ clients do something once in a while
      We should however not break in the middle of a multi-line insert
      if we have binary logging enabled as we don't want other commands
      on this table until all entries has been processed
    */
    if (group_count++ >= max_rows && (row= rows.head()) &&
2598
	(!(row->log_query & using_bin_log)))
unknown's avatar
unknown committed
2599 2600 2601 2602 2603 2604 2605 2606 2607 2608 2609 2610 2611
    {
      group_count=0;
      if (stacked_inserts || tables_in_use)	// Let these wait a while
      {
	if (tables_in_use)
	  pthread_cond_broadcast(&cond_client); // If waiting clients
	thd.proc_info="reschedule";
	pthread_mutex_unlock(&mutex);
	if ((error=table->file->extra(HA_EXTRA_NO_CACHE)))
	{
	  /* This should never happen */
	  table->file->print_error(error,MYF(0));
	  sql_print_error("%s",thd.net.last_error);
2612
          DBUG_PRINT("error", ("HA_EXTRA_NO_CACHE failed in loop"));
unknown's avatar
unknown committed
2613 2614
	  goto err;
	}
unknown's avatar
unknown committed
2615
	query_cache_invalidate3(&thd, table, 1);
unknown's avatar
unknown committed
2616 2617 2618
	if (thr_reschedule_write_lock(*thd.lock->locks))
	{
	  /* This should never happen */
unknown's avatar
unknown committed
2619 2620
	  sql_print_error(ER(ER_DELAYED_CANT_CHANGE_LOCK),
                          table->s->table_name.str);
unknown's avatar
unknown committed
2621
	}
2622 2623
	if (!using_bin_log)
	  table->file->extra(HA_EXTRA_WRITE_CACHE);
unknown's avatar
unknown committed
2624 2625 2626 2627 2628 2629 2630 2631 2632
	pthread_mutex_lock(&mutex);
	thd.proc_info="insert";
      }
      if (tables_in_use)
	pthread_cond_broadcast(&cond_client);	// If waiting clients
    }
  }
  thd.proc_info=0;
  pthread_mutex_unlock(&mutex);
2633

2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647
  /*
    We need to flush the pending event when using row-based
    replication since the flushing normally done in binlog_query() is
    not done last in the statement: for delayed inserts, the insert
    statement is logged *before* all rows are inserted.

    We can flush the pending event without checking the thd->lock
    since the delayed insert *thread* is not inside a stored function
    or trigger.

    TODO: Move the logging to last in the sequence of rows.
   */
  if (thd.current_stmt_binlog_row_based)
    thd.binlog_flush_pending_rows_event(TRUE);
2648

unknown's avatar
unknown committed
2649 2650 2651 2652
  if ((error=table->file->extra(HA_EXTRA_NO_CACHE)))
  {						// This shouldn't happen
    table->file->print_error(error,MYF(0));
    sql_print_error("%s",thd.net.last_error);
2653
    DBUG_PRINT("error", ("HA_EXTRA_NO_CACHE failed after loop"));
unknown's avatar
unknown committed
2654 2655
    goto err;
  }
unknown's avatar
unknown committed
2656
  query_cache_invalidate3(&thd, table, 1);
unknown's avatar
unknown committed
2657 2658 2659 2660
  pthread_mutex_lock(&mutex);
  DBUG_RETURN(0);

 err:
2661 2662 2663
#ifndef DBUG_OFF
  max_rows= 0;                                  // For DBUG output
#endif
2664 2665 2666 2667 2668 2669
  /* Remove all not used rows */
  while ((row=rows.get()))
  {
    delete row;
    thread_safe_increment(delayed_insert_errors,&LOCK_delayed_status);
    stacked_inserts--;
2670 2671 2672
#ifndef DBUG_OFF
    max_rows++;
#endif
2673
  }
2674
  DBUG_PRINT("error", ("dropped %lu rows after an error", max_rows));
unknown's avatar
unknown committed
2675 2676 2677 2678
  thread_safe_increment(delayed_insert_errors, &LOCK_delayed_status);
  pthread_mutex_lock(&mutex);
  DBUG_RETURN(1);
}
2679
#endif /* EMBEDDED_LIBRARY */
unknown's avatar
unknown committed
2680 2681

/***************************************************************************
unknown's avatar
unknown committed
2682
  Store records in INSERT ... SELECT *
unknown's avatar
unknown committed
2683 2684
***************************************************************************/

unknown's avatar
VIEW  
unknown committed
2685 2686 2687 2688 2689 2690 2691 2692 2693

/*
  make insert specific preparation and checks after opening tables

  SYNOPSIS
    mysql_insert_select_prepare()
    thd         thread handler

  RETURN
unknown's avatar
unknown committed
2694 2695
    FALSE OK
    TRUE  Error
unknown's avatar
VIEW  
unknown committed
2696 2697
*/

unknown's avatar
unknown committed
2698
bool mysql_insert_select_prepare(THD *thd)
unknown's avatar
VIEW  
unknown committed
2699 2700
{
  LEX *lex= thd->lex;
unknown's avatar
unknown committed
2701
  SELECT_LEX *select_lex= &lex->select_lex;
unknown's avatar
unknown committed
2702
  TABLE_LIST *first_select_leaf_table;
unknown's avatar
VIEW  
unknown committed
2703
  DBUG_ENTER("mysql_insert_select_prepare");
unknown's avatar
unknown committed
2704

unknown's avatar
unknown committed
2705 2706
  /*
    SELECT_LEX do not belong to INSERT statement, so we can't add WHERE
unknown's avatar
unknown committed
2707
    clause if table is VIEW
unknown's avatar
unknown committed
2708
  */
unknown's avatar
unknown committed
2709
  
unknown's avatar
unknown committed
2710 2711 2712 2713
  if (mysql_prepare_insert(thd, lex->query_tables,
                           lex->query_tables->table, lex->field_list, 0,
                           lex->update_list, lex->value_list,
                           lex->duplicates,
unknown's avatar
unknown committed
2714
                           &select_lex->where, TRUE, FALSE, FALSE))
unknown's avatar
unknown committed
2715
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
2716

2717 2718 2719 2720
  /*
    exclude first table from leaf tables list, because it belong to
    INSERT
  */
unknown's avatar
unknown committed
2721 2722
  DBUG_ASSERT(select_lex->leaf_tables != 0);
  lex->leaf_tables_insert= select_lex->leaf_tables;
2723
  /* skip all leaf tables belonged to view where we are insert */
2724
  for (first_select_leaf_table= select_lex->leaf_tables->next_leaf;
2725 2726 2727 2728
       first_select_leaf_table &&
       first_select_leaf_table->belong_to_view &&
       first_select_leaf_table->belong_to_view ==
       lex->leaf_tables_insert->belong_to_view;
2729
       first_select_leaf_table= first_select_leaf_table->next_leaf)
2730
  {}
unknown's avatar
unknown committed
2731
  select_lex->leaf_tables= first_select_leaf_table;
unknown's avatar
unknown committed
2732
  DBUG_RETURN(FALSE);
unknown's avatar
VIEW  
unknown committed
2733 2734 2735
}


2736
select_insert::select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
unknown's avatar
unknown committed
2737
                             List<Item> *fields_par,
unknown's avatar
unknown committed
2738 2739
                             List<Item> *update_fields,
                             List<Item> *update_values,
unknown's avatar
unknown committed
2740
                             enum_duplicates duplic,
2741 2742
                             bool ignore_check_option_errors)
  :table_list(table_list_par), table(table_par), fields(fields_par),
2743
   autoinc_value_of_last_inserted_row(0),
2744 2745 2746
   insert_into_view(table_list_par && table_list_par->view != 0)
{
  bzero((char*) &info,sizeof(info));
unknown's avatar
unknown committed
2747 2748 2749 2750
  info.handle_duplicates= duplic;
  info.ignore= ignore_check_option_errors;
  info.update_fields= update_fields;
  info.update_values= update_values;
2751
  if (table_list_par)
unknown's avatar
unknown committed
2752
    info.view= (table_list_par->view ? table_list_par : 0);
2753 2754 2755
}


unknown's avatar
unknown committed
2756
int
2757
select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
unknown's avatar
unknown committed
2758
{
2759
  LEX *lex= thd->lex;
2760
  int res;
2761
  table_map map= 0;
2762
  SELECT_LEX *lex_current_select_save= lex->current_select;
unknown's avatar
unknown committed
2763 2764
  DBUG_ENTER("select_insert::prepare");

2765
  unit= u;
2766

2767 2768 2769 2770 2771 2772
  /*
    Since table in which we are going to insert is added to the first
    select, LEX::current_select should point to the first select while
    we are fixing fields from insert list.
  */
  lex->current_select= &lex->select_lex;
2773
  res= check_insert_fields(thd, table_list, *fields, values,
2774
                           !insert_into_view, &map) ||
2775
       setup_fields(thd, 0, values, MARK_COLUMNS_READ, 0, 0);
2776

unknown's avatar
unknown committed
2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788
  if (!res && fields->elements)
  {
    bool saved_abort_on_warning= thd->abort_on_warning;
    thd->abort_on_warning= !info.ignore && (thd->variables.sql_mode &
                                            (MODE_STRICT_TRANS_TABLES |
                                             MODE_STRICT_ALL_TABLES));
    res= check_that_all_fields_are_given_values(thd, table_list->table, 
                                                table_list);
    thd->abort_on_warning= saved_abort_on_warning;
  }

  if (info.handle_duplicates == DUP_UPDATE && !res)
2789
  {
2790
    Name_resolution_context *context= &lex->select_lex.context;
2791 2792 2793 2794
    Name_resolution_context_state ctx_state;

    /* Save the state of the current name resolution context. */
    ctx_state.save_state(context, table_list);
2795 2796

    /* Perform name resolution only in the first table - 'table_list'. */
2797
    table_list->next_local= 0;
2798 2799
    context->resolve_in_table_list_only(table_list);

2800
    lex->select_lex.no_wrap_view_item= TRUE;
2801
    res= res || check_update_fields(thd, context->table_list,
2802
                                    *info.update_fields, &map);
2803 2804
    lex->select_lex.no_wrap_view_item= FALSE;
    /*
2805 2806 2807
      When we are not using GROUP BY and there are no ungrouped aggregate functions 
      we can refer to other tables in the ON DUPLICATE KEY part.
      We use next_name_resolution_table descructively, so check it first (views?)
2808
    */
2809 2810 2811 2812 2813 2814 2815 2816
    DBUG_ASSERT (!table_list->next_name_resolution_table);
    if (lex->select_lex.group_list.elements == 0 &&
        !lex->select_lex.with_sum_func)
      /*
        We must make a single context out of the two separate name resolution contexts :
        the INSERT table and the tables in the SELECT part of INSERT ... SELECT.
        To do that we must concatenate the two lists
      */  
unknown's avatar
unknown committed
2817 2818
      table_list->next_name_resolution_table= 
        ctx_state.get_first_name_resolution_table();
2819

unknown's avatar
unknown committed
2820 2821
    res= res || setup_fields(thd, 0, *info.update_values,
                             MARK_COLUMNS_READ, 0, 0);
2822
    if (!res)
2823
    {
2824 2825 2826 2827 2828 2829 2830 2831
      /*
        Traverse the update values list and substitute fields from the
        select for references (Item_ref objects) to them. This is done in
        order to get correct values from those fields when the select
        employs a temporary table.
      */
      List_iterator<Item> li(*info.update_values);
      Item *item;
2832

2833 2834 2835
      while ((item= li++))
      {
        item->transform(&Item::update_value_transformer,
2836
                        (uchar*)lex->current_select);
2837
      }
2838 2839 2840
    }

    /* Restore the current context. */
2841
    ctx_state.restore_state(context, table_list);
2842
  }
2843

2844 2845
  lex->current_select= lex_current_select_save;
  if (res)
unknown's avatar
unknown committed
2846
    DBUG_RETURN(1);
2847 2848 2849 2850 2851 2852 2853 2854 2855 2856
  /*
    if it is INSERT into join view then check_insert_fields already found
    real table for insert
  */
  table= table_list->table;

  /*
    Is table which we are changing used somewhere in other parts of
    query
  */
2857
  if (!(lex->current_select->options & OPTION_BUFFER_RESULT) &&
2858
      unique_table(thd, table_list, table_list->next_global, 0))
2859 2860
  {
    /* Using same table for INSERT and SELECT */
2861 2862
    lex->current_select->options|= OPTION_BUFFER_RESULT;
    lex->current_select->join->select_options|= OPTION_BUFFER_RESULT;
2863
  }
2864
  else if (!thd->prelocked_mode)
2865 2866 2867 2868 2869 2870 2871
  {
    /*
      We must not yet prepare the result table if it is the same as one of the 
      source tables (INSERT SELECT). The preparation may disable 
      indexes on the result table, which may be used during the select, if it
      is the same table (Bug #6034). Do the preparation after the select phase
      in select_insert::prepare2().
2872 2873
      We won't start bulk inserts at all if this statement uses functions or
      should invoke triggers since they may access to the same table too.
2874
    */
2875
    table->file->ha_start_bulk_insert((ha_rows) 0);
2876
  }
2877
  restore_record(table,s->default_values);		// Get empty record
unknown's avatar
unknown committed
2878
  table->next_number_field=table->found_next_number_field;
2879 2880 2881 2882 2883 2884 2885 2886 2887

#ifdef HAVE_REPLICATION
  if (thd->slave_thread &&
      (info.handle_duplicates == DUP_UPDATE) &&
      (table->next_number_field != NULL) &&
      rpl_master_has_bug(&active_mi->rli, 24432))
    DBUG_RETURN(1);
#endif

unknown's avatar
unknown committed
2888
  thd->cuted_fields=0;
unknown's avatar
unknown committed
2889 2890
  if (info.ignore || info.handle_duplicates != DUP_ERROR)
    table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
2891 2892 2893
  if (info.handle_duplicates == DUP_REPLACE &&
      (!table->triggers || !table->triggers->has_delete_triggers()))
    table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
unknown's avatar
unknown committed
2894
  thd->no_trans_update.stmt= FALSE;
unknown's avatar
unknown committed
2895
  thd->abort_on_warning= (!info.ignore &&
unknown's avatar
unknown committed
2896 2897 2898
                          (thd->variables.sql_mode &
                           (MODE_STRICT_TRANS_TABLES |
                            MODE_STRICT_ALL_TABLES)));
unknown's avatar
unknown committed
2899
  res= (table_list->prepare_where(thd, 0, TRUE) ||
2900
        table_list->prepare_check_option(thd));
2901 2902

  if (!res)
unknown's avatar
unknown committed
2903
     prepare_triggers_for_insert_stmt(table);
2904

2905
  DBUG_RETURN(res);
unknown's avatar
unknown committed
2906 2907
}

2908

2909 2910 2911 2912 2913 2914 2915 2916 2917 2918 2919 2920 2921 2922 2923 2924 2925 2926 2927
/*
  Finish the preparation of the result table.

  SYNOPSIS
    select_insert::prepare2()
    void

  DESCRIPTION
    If the result table is the same as one of the source tables (INSERT SELECT),
    the result table is not finally prepared at the join prepair phase.
    Do the final preparation now.
		       
  RETURN
    0   OK
*/

int select_insert::prepare2(void)
{
  DBUG_ENTER("select_insert::prepare2");
2928 2929
  if (thd->lex->current_select->options & OPTION_BUFFER_RESULT &&
      !thd->prelocked_mode)
2930
    table->file->ha_start_bulk_insert((ha_rows) 0);
unknown's avatar
unknown committed
2931
  DBUG_RETURN(0);
2932 2933 2934
}


2935 2936 2937 2938 2939 2940
void select_insert::cleanup()
{
  /* select_insert/select_create are never re-used in prepared statement */
  DBUG_ASSERT(0);
}

unknown's avatar
unknown committed
2941 2942
select_insert::~select_insert()
{
2943
  DBUG_ENTER("~select_insert");
unknown's avatar
unknown committed
2944 2945 2946
  if (table)
  {
    table->next_number_field=0;
2947
    table->auto_increment_field_not_null= FALSE;
2948
    table->file->ha_reset();
unknown's avatar
unknown committed
2949
  }
2950
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
unknown's avatar
unknown committed
2951
  thd->abort_on_warning= 0;
2952
  DBUG_VOID_RETURN;
unknown's avatar
unknown committed
2953 2954 2955 2956 2957
}


bool select_insert::send_data(List<Item> &values)
{
2958
  DBUG_ENTER("select_insert::send_data");
unknown's avatar
unknown committed
2959
  bool error=0;
2960

2961
  if (unit->offset_limit_cnt)
unknown's avatar
unknown committed
2962
  {						// using limit offset,count
2963
    unit->offset_limit_cnt--;
2964
    DBUG_RETURN(0);
unknown's avatar
unknown committed
2965
  }
unknown's avatar
unknown committed
2966

unknown's avatar
unknown committed
2967
  thd->count_cuted_fields= CHECK_FIELD_WARN;	// Calculate cuted fields
unknown's avatar
unknown committed
2968 2969
  store_values(values);
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
unknown's avatar
unknown committed
2970 2971
  if (thd->net.report_error)
    DBUG_RETURN(1);
unknown's avatar
unknown committed
2972 2973
  if (table_list)                               // Not CREATE ... SELECT
  {
unknown's avatar
unknown committed
2974
    switch (table_list->view_check_option(thd, info.ignore)) {
unknown's avatar
unknown committed
2975 2976 2977 2978 2979
    case VIEW_CHECK_SKIP:
      DBUG_RETURN(0);
    case VIEW_CHECK_ERROR:
      DBUG_RETURN(1);
    }
unknown's avatar
unknown committed
2980
  }
2981

2982
  error= write_record(thd, table, &info);
2983 2984
    
  if (!error)
unknown's avatar
unknown committed
2985
  {
unknown's avatar
unknown committed
2986
    if (table->triggers || info.handle_duplicates == DUP_UPDATE)
unknown's avatar
unknown committed
2987 2988
    {
      /*
unknown's avatar
unknown committed
2989 2990 2991
        Restore fields of the record since it is possible that they were
        changed by ON DUPLICATE KEY UPDATE clause.
    
unknown's avatar
unknown committed
2992 2993 2994 2995 2996 2997 2998 2999
        If triggers exist then whey can modify some fields which were not
        originally touched by INSERT ... SELECT, so we have to restore
        their original values for the next row.
      */
      restore_record(table, s->default_values);
    }
    if (table->next_number_field)
    {
3000 3001 3002 3003 3004 3005 3006
      /*
        If no value has been autogenerated so far, we need to remember the
        value we just saw, we may need to send it to client in the end.
      */
      if (thd->first_successful_insert_id_in_cur_stmt == 0) // optimization
        autoinc_value_of_last_inserted_row= 
          table->next_number_field->val_int();
unknown's avatar
unknown committed
3007 3008 3009 3010 3011 3012
      /*
        Clear auto-increment field for the next record, if triggers are used
        we will clear it twice, but this should be cheap.
      */
      table->next_number_field->reset();
    }
unknown's avatar
unknown committed
3013
  }
unknown's avatar
unknown committed
3014
  DBUG_RETURN(error);
unknown's avatar
unknown committed
3015 3016 3017
}


unknown's avatar
unknown committed
3018 3019 3020
void select_insert::store_values(List<Item> &values)
{
  if (fields->elements)
unknown's avatar
unknown committed
3021 3022
    fill_record_n_invoke_before_triggers(thd, *fields, values, 1,
                                         table->triggers, TRG_EVENT_INSERT);
unknown's avatar
unknown committed
3023
  else
unknown's avatar
unknown committed
3024 3025
    fill_record_n_invoke_before_triggers(thd, table->field, values, 1,
                                         table->triggers, TRG_EVENT_INSERT);
unknown's avatar
unknown committed
3026 3027
}

unknown's avatar
unknown committed
3028 3029
void select_insert::send_error(uint errcode,const char *err)
{
unknown's avatar
unknown committed
3030 3031
  DBUG_ENTER("select_insert::send_error");

3032
  my_message(errcode, err, MYF(0));
3033

unknown's avatar
unknown committed
3034
  DBUG_VOID_RETURN;
unknown's avatar
unknown committed
3035 3036 3037 3038 3039
}


bool select_insert::send_eof()
{
3040
  int error;
3041
  bool const trans_table= table->file->has_transactions();
3042
  ulonglong id;
unknown's avatar
unknown committed
3043
  DBUG_ENTER("select_insert::send_eof");
3044 3045
  DBUG_PRINT("enter", ("trans_table=%d, table_type='%s'",
                       trans_table, table->file->table_type()));
unknown's avatar
unknown committed
3046

3047
  error= (!thd->prelocked_mode) ? table->file->ha_end_bulk_insert():0;
unknown's avatar
unknown committed
3048
  table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
3049
  table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
3050

unknown's avatar
unknown committed
3051
  if (info.copied || info.deleted || info.updated)
3052 3053 3054 3055 3056
  {
    /*
      We must invalidate the table in the query cache before binlog writing
      and ha_autocommit_or_rollback.
    */
unknown's avatar
unknown committed
3057
    query_cache_invalidate3(thd, table, 1);
3058 3059 3060 3061 3062 3063 3064
    /*
      Mark that we have done permanent changes if all of the below is true
      - Table doesn't support transactions
      - It's a normal (not temporary) table. (Changes to temporary tables
        are not logged in RBR)
      - We are using statement based replication
    */
3065 3066
    if (!trans_table &&
        (!table->s->tmp_table || !thd->current_stmt_binlog_row_based))
3067
      thd->no_trans_update.all= TRUE;
3068
   }
unknown's avatar
unknown committed
3069

3070 3071 3072 3073 3074 3075
  /*
    Write to binlog before commiting transaction.  No statement will
    be written by the binlog_query() below in RBR mode.  All the
    events are in the transaction cache and will be written when
    ha_autocommit_or_rollback() is issued below.
  */
3076 3077
  if (mysql_bin_log.is_open())
  {
unknown's avatar
unknown committed
3078 3079
    if (!error)
      thd->clear_error();
3080 3081
    thd->binlog_query(THD::ROW_QUERY_TYPE,
                      thd->query, thd->query_length,
3082
                      trans_table, FALSE);
3083
  }
3084 3085 3086 3087 3088 3089 3090 3091
  /*
    We will call ha_autocommit_or_rollback() also for
    non-transactional tables under row-based replication: there might
    be events in the binary logs transaction, and we need to write
    them to the binary log.
   */
  if (trans_table || thd->current_stmt_binlog_row_based)
  {
3092
    int error2= ha_autocommit_or_rollback(thd, error);
3093
    if (error2 && !error)
3094
      error= error2;
3095
  }
3096
  table->file->ha_release_auto_increment();
3097

3098
  if (error)
unknown's avatar
unknown committed
3099 3100
  {
    table->file->print_error(error,MYF(0));
unknown's avatar
unknown committed
3101
    DBUG_RETURN(1);
unknown's avatar
unknown committed
3102
  }
unknown's avatar
unknown committed
3103
  char buff[160];
3104
  if (info.ignore)
unknown's avatar
unknown committed
3105 3106
    sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
	    (ulong) (info.records - info.copied), (ulong) thd->cuted_fields);
unknown's avatar
unknown committed
3107
  else
unknown's avatar
unknown committed
3108
    sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
unknown's avatar
unknown committed
3109
	    (ulong) (info.deleted+info.updated), (ulong) thd->cuted_fields);
3110 3111 3112
  thd->row_count_func= info.copied + info.deleted +
                       ((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
                        info.touched : info.updated);
3113 3114 3115 3116 3117 3118 3119

  id= (thd->first_successful_insert_id_in_cur_stmt > 0) ?
    thd->first_successful_insert_id_in_cur_stmt :
    (thd->arg_of_last_insert_id_function ?
     thd->first_successful_insert_id_in_prev_stmt :
     (info.copied ? autoinc_value_of_last_inserted_row : 0));
  ::send_ok(thd, (ulong) thd->row_count_func, id, buff);
unknown's avatar
unknown committed
3120
  DBUG_RETURN(0);
unknown's avatar
unknown committed
3121 3122
}

3123 3124 3125 3126 3127 3128 3129 3130 3131 3132 3133 3134 3135 3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162 3163 3164 3165 3166 3167 3168 3169 3170 3171 3172 3173 3174 3175
void select_insert::abort() {

  DBUG_ENTER("select_insert::abort");
  /*
    If the creation of the table failed (due to a syntax error, for
    example), no table will have been opened and therefore 'table'
    will be NULL. In that case, we still need to execute the rollback
    and the end of the function.
   */
  if (table)
  {
    /*
      If we are not in prelocked mode, we end the bulk insert started
      before.
    */
    if (!thd->prelocked_mode)
      table->file->ha_end_bulk_insert();

    /*
      If at least one row has been inserted/modified and will stay in
      the table (the table doesn't have transactions) we must write to
      the binlog (and the error code will make the slave stop).

      For many errors (example: we got a duplicate key error while
      inserting into a MyISAM table), no row will be added to the table,
      so passing the error to the slave will not help since there will
      be an error code mismatch (the inserts will succeed on the slave
      with no error).

      If table creation failed, the number of rows modified will also be
      zero, so no check for that is made.
    */
    if (info.copied || info.deleted || info.updated)
    {
      DBUG_ASSERT(table != NULL);
      if (!table->file->has_transactions())
      {
        if (mysql_bin_log.is_open())
          thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query, thd->query_length,
                            table->file->has_transactions(), FALSE);
        if (!thd->current_stmt_binlog_row_based && !table->s->tmp_table &&
            !can_rollback_data())
          thd->no_trans_update.all= TRUE;
        query_cache_invalidate3(thd, table, 1);
      }
    }
    table->file->ha_release_auto_increment();
  }

  ha_rollback_stmt(thd);
  DBUG_VOID_RETURN;
}

unknown's avatar
unknown committed
3176 3177

/***************************************************************************
unknown's avatar
unknown committed
3178
  CREATE TABLE (SELECT) ...
unknown's avatar
unknown committed
3179 3180
***************************************************************************/

3181
/*
unknown's avatar
unknown committed
3182 3183
  Create table from lists of fields and items (or just return TABLE
  object for pre-opened existing table).
3184 3185 3186 3187 3188 3189 3190 3191

  SYNOPSIS
    create_table_from_items()
      thd          in     Thread object
      create_info  in     Create information (like MAX_ROWS, ENGINE or
                          temporary table flag)
      create_table in     Pointer to TABLE_LIST object providing database
                          and name for table to be created or to be open
3192 3193
      alter_info   in/out Initial list of columns and indexes for the table
                          to be created
3194 3195
      items        in     List of items which should be used to produce rest
                          of fields for the table (corresponding fields will
3196
                          be added to the end of alter_info->create_list)
3197
      lock         out    Pointer to the MYSQL_LOCK object for table created
unknown's avatar
unknown committed
3198 3199 3200 3201
                          (or open temporary table) will be returned in this
                          parameter. Since this table is not included in
                          THD::lock caller is responsible for explicitly
                          unlocking this table.
unknown's avatar
unknown committed
3202
      hooks
3203 3204

  NOTES
unknown's avatar
unknown committed
3205 3206 3207 3208 3209 3210 3211 3212 3213 3214 3215 3216
    This function behaves differently for base and temporary tables:
    - For base table we assume that either table exists and was pre-opened
      and locked at open_and_lock_tables() stage (and in this case we just
      emit error or warning and return pre-opened TABLE object) or special
      placeholder was put in table cache that guarantees that this table
      won't be created or opened until the placeholder will be removed
      (so there is an exclusive lock on this table).
    - We don't pre-open existing temporary table, instead we either open
      or create and then open table in this function.

    Since this function contains some logic specific to CREATE TABLE ...
    SELECT it should be changed before it can be used in other contexts.
3217 3218 3219 3220 3221 3222 3223 3224

  RETURN VALUES
    non-zero  Pointer to TABLE object for table created or opened
    0         Error
*/

static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
                                      TABLE_LIST *create_table,
3225
                                      Alter_info *alter_info,
unknown's avatar
unknown committed
3226 3227 3228
                                      List<Item> *items,
                                      MYSQL_LOCK **lock,
                                      TABLEOP_HOOKS *hooks)
3229
{
unknown's avatar
unknown committed
3230
  TABLE tmp_table;		// Used during 'Create_field()'
unknown's avatar
unknown committed
3231
  TABLE_SHARE share;
3232 3233 3234 3235 3236 3237 3238 3239 3240
  TABLE *table= 0;
  uint select_field_count= items->elements;
  /* Add selected items to field list */
  List_iterator_fast<Item> it(*items);
  Item *item;
  Field *tmp_field;
  bool not_used;
  DBUG_ENTER("create_table_from_items");

unknown's avatar
unknown committed
3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259
  DBUG_EXECUTE_IF("sleep_create_select_before_check_if_exists", my_sleep(6000000););

  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
      create_table->table->db_stat)
  {
    /* Table already exists and was open at open_and_lock_tables() stage. */
    if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
    {
      create_info->table_existed= 1;		// Mark that table existed
      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                          ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                          create_table->table_name);
      DBUG_RETURN(create_table->table);
    }

    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
    DBUG_RETURN(0);
  }

3260 3261
  tmp_table.alias= 0;
  tmp_table.timestamp_field= 0;
unknown's avatar
unknown committed
3262 3263 3264
  tmp_table.s= &share;
  init_tmp_table_share(&share, "", 0, "", "");

3265 3266
  tmp_table.s->db_create_options=0;
  tmp_table.s->blob_ptr_size= portable_sizeof_char_ptr;
unknown's avatar
unknown committed
3267
  tmp_table.s->db_low_byte_first= 
3268 3269
        test(create_info->db_type == myisam_hton ||
             create_info->db_type == heap_hton);
3270 3271 3272 3273
  tmp_table.null_row=tmp_table.maybe_null=0;

  while ((item=it++))
  {
unknown's avatar
unknown committed
3274
    Create_field *cr_field;
3275
    Field *field, *def_field;
3276
    if (item->type() == Item::FUNC_ITEM)
3277
      field= item->tmp_table_field(&tmp_table);
3278
    else
3279 3280 3281
      field= create_tmp_field(thd, &tmp_table, item, item->type(),
                              (Item ***) 0, &tmp_field, &def_field, 0, 0, 0, 0,
                              0);
3282
    if (!field ||
unknown's avatar
unknown committed
3283
	!(cr_field=new Create_field(field,(item->type() == Item::FIELD_ITEM ?
3284 3285 3286 3287 3288
					   ((Item_field *)item)->field :
					   (Field*) 0))))
      DBUG_RETURN(0);
    if (item->maybe_null)
      cr_field->flags &= ~NOT_NULL_FLAG;
3289
    alter_info->create_list.push_back(cr_field);
3290
  }
unknown's avatar
unknown committed
3291 3292 3293

  DBUG_EXECUTE_IF("sleep_create_select_before_create", my_sleep(6000000););

3294
  /*
unknown's avatar
unknown committed
3295 3296 3297 3298 3299
    Create and lock table.

    Note that we either creating (or opening existing) temporary table or
    creating base table on which name we have exclusive lock. So code below
    should not cause deadlocks or races.
3300 3301 3302 3303 3304 3305 3306 3307 3308 3309 3310 3311

    We don't log the statement, it will be logged later.

    If this is a HEAP table, the automatic DELETE FROM which is written to the
    binlog when a HEAP table is opened for the first time since startup, must
    not be written: 1) it would be wrong (imagine we're in CREATE SELECT: we
    don't want to delete from it) 2) it would be written before the CREATE
    TABLE, which is a wrong order. So we keep binary logging disabled when we
    open_table().
  */
  {
    tmp_disable_binlog(thd);
unknown's avatar
unknown committed
3312 3313
    if (!mysql_create_table_no_lock(thd, create_table->db,
                                    create_table->table_name,
3314 3315
                                    create_info, alter_info, 0,
                                    select_field_count))
3316
    {
unknown's avatar
unknown committed
3317 3318 3319 3320 3321 3322 3323 3324 3325 3326 3327 3328 3329 3330 3331 3332 3333 3334 3335 3336 3337 3338 3339 3340 3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356 3357
      if (create_info->table_existed &&
          !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
      {
        /*
          This means that someone created table underneath server
          or it was created via different mysqld front-end to the
          cluster. We don't have much options but throw an error.
        */
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
        DBUG_RETURN(0);
      }

      DBUG_EXECUTE_IF("sleep_create_select_before_open", my_sleep(6000000););

      if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
      {
        VOID(pthread_mutex_lock(&LOCK_open));
        if (reopen_name_locked_table(thd, create_table, FALSE))
        {
          quick_rm_table(create_info->db_type, create_table->db,
                         table_case_name(create_info, create_table->table_name),
                         0);
        }
        else
          table= create_table->table;
        VOID(pthread_mutex_unlock(&LOCK_open));
      }
      else
      {
        if (!(table= open_table(thd, create_table, thd->mem_root, (bool*) 0,
                                MYSQL_OPEN_TEMPORARY_ONLY)) &&
            !create_info->table_existed)
        {
          /*
            This shouldn't happen as creation of temporary table should make
            it preparable for open. But let us do close_temporary_table() here
            just in case.
          */
          close_temporary_table(thd, create_table);
        }
      }
3358 3359 3360 3361 3362 3363
    }
    reenable_binlog(thd);
    if (!table)                                   // open failed
      DBUG_RETURN(0);
  }

unknown's avatar
unknown committed
3364 3365
  DBUG_EXECUTE_IF("sleep_create_select_before_lock", my_sleep(6000000););

3366
  table->reginfo.lock_type=TL_WRITE;
unknown's avatar
unknown committed
3367
  hooks->prelock(&table, 1);                    // Call prelock hooks
3368
  if (! ((*lock)= mysql_lock_tables(thd, &table, 1,
3369 3370
                                    MYSQL_LOCK_IGNORE_FLUSH, &not_used)) ||
        hooks->postlock(&table, 1))
3371
  {
3372 3373 3374 3375 3376 3377
    if (*lock)
    {
      mysql_unlock_tables(thd, *lock);
      *lock= 0;
    }

unknown's avatar
unknown committed
3378 3379
    if (!create_info->table_existed)
      drop_open_table(thd, table, create_table->db, create_table->table_name);
3380 3381 3382 3383 3384 3385
    DBUG_RETURN(0);
  }
  DBUG_RETURN(table);
}


unknown's avatar
unknown committed
3386
int
3387
select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
3388 3389
{
  DBUG_ENTER("select_create::prepare");
3390

3391
  TABLEOP_HOOKS *hook_ptr= NULL;
3392 3393 3394 3395 3396 3397 3398 3399 3400 3401 3402 3403 3404 3405 3406 3407 3408 3409
  /*
    For row-based replication, the CREATE-SELECT statement is written
    in two pieces: the first one contain the CREATE TABLE statement
    necessary to create the table and the second part contain the rows
    that should go into the table.

    For non-temporary tables, the start of the CREATE-SELECT
    implicitly commits the previous transaction, and all events
    forming the statement will be stored the transaction cache. At end
    of the statement, the entire statement is committed as a
    transaction, and all events are written to the binary log.

    On the master, the table is locked for the duration of the
    statement, but since the CREATE part is replicated as a simple
    statement, there is no way to lock the table for accesses on the
    slave.  Hence, we have to hold on to the CREATE part of the
    statement until the statement has finished.
   */
3410 3411
  class MY_HOOKS : public TABLEOP_HOOKS {
  public:
3412 3413 3414 3415 3416 3417
    MY_HOOKS(select_create *x, TABLE_LIST *create_table,
             TABLE_LIST *select_tables)
      : ptr(x), all_tables(*create_table)
      {
        all_tables.next_global= select_tables;
      }
3418 3419

  private:
3420
    virtual int do_postlock(TABLE **tables, uint count)
3421
    {
3422 3423 3424 3425
      THD *thd= const_cast<THD*>(ptr->get_thd());
      if (int error= decide_logging_format(thd, &all_tables))
        return error;

3426
      TABLE const *const table = *tables;
3427
      if (thd->current_stmt_binlog_row_based  &&
3428
          !table->s->tmp_table &&
3429 3430 3431 3432
          !ptr->get_create_info()->table_existed)
      {
        ptr->binlog_show_create_table(tables, count);
      }
3433
      return 0;
3434 3435 3436
    }

    select_create *ptr;
3437
    TABLE_LIST all_tables;
3438 3439
  };

3440
  MY_HOOKS hooks(this, create_table, select_tables);
3441
  hook_ptr= &hooks;
3442

3443
  unit= u;
3444 3445

  /*
3446 3447 3448
    Start a statement transaction before the create if we are using
    row-based replication for the statement.  If we are creating a
    temporary table, we need to start a statement transaction.
3449 3450 3451 3452 3453 3454 3455
  */
  if ((thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) == 0 &&
      thd->current_stmt_binlog_row_based)
  {
    thd->binlog_start_trans_and_stmt();
  }

3456
  if (!(table= create_table_from_items(thd, create_info, create_table,
3457
                                       alter_info, &values,
3458
                                       &thd->extra_lock, hook_ptr)))
unknown's avatar
unknown committed
3459 3460
    DBUG_RETURN(-1);				// abort() deletes table

3461
  if (table->s->fields < values.elements)
unknown's avatar
unknown committed
3462
  {
3463
    my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1);
unknown's avatar
unknown committed
3464 3465 3466
    DBUG_RETURN(-1);
  }

3467
 /* First field to copy */
unknown's avatar
unknown committed
3468 3469 3470 3471
  field= table->field+table->s->fields - values.elements;

  /* Mark all fields that are given values */
  for (Field **f= field ; *f ; f++)
3472
    bitmap_set_bit(table->write_set, (*f)->field_index);
unknown's avatar
unknown committed
3473

3474
  /* Don't set timestamp if used */
3475
  table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
unknown's avatar
unknown committed
3476 3477
  table->next_number_field=table->found_next_number_field;

3478
  restore_record(table,s->default_values);      // Get empty record
unknown's avatar
unknown committed
3479
  thd->cuted_fields=0;
unknown's avatar
unknown committed
3480
  if (info.ignore || info.handle_duplicates != DUP_ERROR)
unknown's avatar
unknown committed
3481
    table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
3482 3483 3484
  if (info.handle_duplicates == DUP_REPLACE &&
      (!table->triggers || !table->triggers->has_delete_triggers()))
    table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
3485
  if (!thd->prelocked_mode)
3486
    table->file->ha_start_bulk_insert((ha_rows) 0);
3487
  thd->no_trans_update.stmt= FALSE;
unknown's avatar
unknown committed
3488
  thd->abort_on_warning= (!info.ignore &&
unknown's avatar
unknown committed
3489 3490 3491
                          (thd->variables.sql_mode &
                           (MODE_STRICT_TRANS_TABLES |
                            MODE_STRICT_ALL_TABLES)));
3492 3493 3494
  if (check_that_all_fields_are_given_values(thd, table, table_list))
    DBUG_RETURN(1);
  table->mark_columns_needed_for_insert();
unknown's avatar
unknown committed
3495
  table->file->extra(HA_EXTRA_WRITE_CACHE);
3496
  DBUG_RETURN(0);
unknown's avatar
unknown committed
3497 3498
}

3499
void
3500
select_create::binlog_show_create_table(TABLE **tables, uint count)
3501 3502 3503 3504 3505 3506 3507 3508 3509 3510 3511 3512 3513 3514 3515 3516 3517
{
  /*
    Note 1: In RBR mode, we generate a CREATE TABLE statement for the
    created table by calling store_create_info() (behaves as SHOW
    CREATE TABLE).  In the event of an error, nothing should be
    written to the binary log, even if the table is non-transactional;
    therefore we pretend that the generated CREATE TABLE statement is
    for a transactional table.  The event will then be put in the
    transaction cache, and any subsequent events (e.g., table-map
    events and binrow events) will also be put there.  We can then use
    ha_autocommit_or_rollback() to either throw away the entire
    kaboodle of events, or write them to the binary log.

    We write the CREATE TABLE statement here and not in prepare()
    since there potentially are sub-selects or accesses to information
    schema that will do a close_thread_tables(), destroying the
    statement transaction cache.
3518
  */
3519
  DBUG_ASSERT(thd->current_stmt_binlog_row_based);
3520
  DBUG_ASSERT(tables && *tables && count > 0);
3521 3522 3523

  char buf[2048];
  String query(buf, sizeof(buf), system_charset_info);
3524
  int result;
3525
  TABLE_LIST table_list;
3526

3527 3528
  memset(&table_list, 0, sizeof(table_list));
  table_list.table = *tables;
3529
  query.length(0);      // Have to zero it since constructor doesn't
3530

3531
  result= store_create_info(thd, &table_list, &query, create_info);
3532
  DBUG_ASSERT(result == 0); /* store_create_info() always return 0 */
3533

3534 3535 3536 3537 3538 3539
  thd->binlog_query(THD::STMT_QUERY_TYPE,
                    query.ptr(), query.length(),
                    /* is_trans */ TRUE,
                    /* suppress_use */ FALSE);
}

unknown's avatar
unknown committed
3540
void select_create::store_values(List<Item> &values)
unknown's avatar
unknown committed
3541
{
unknown's avatar
unknown committed
3542 3543
  fill_record_n_invoke_before_triggers(thd, field, values, 1,
                                       table->triggers, TRG_EVENT_INSERT);
unknown's avatar
unknown committed
3544 3545
}

unknown's avatar
unknown committed
3546

3547 3548
void select_create::send_error(uint errcode,const char *err)
{
3549 3550 3551 3552 3553 3554 3555
  DBUG_ENTER("select_create::send_error");

  DBUG_PRINT("info",
             ("Current statement %s row-based",
              thd->current_stmt_binlog_row_based ? "is" : "is NOT"));
  DBUG_PRINT("info",
             ("Current table (at 0x%lu) %s a temporary (or non-existant) table",
3556
              (ulong) table,
3557 3558 3559 3560 3561
              table && !table->s->tmp_table ? "is NOT" : "is"));
  DBUG_PRINT("info",
             ("Table %s prior to executing this statement",
              get_create_info()->table_existed ? "existed" : "did not exist"));

3562
  /*
3563 3564 3565 3566 3567 3568 3569 3570 3571
    This will execute any rollbacks that are necessary before writing
    the transcation cache.

    We disable the binary log since nothing should be written to the
    binary log.  This disabling is important, since we potentially do
    a "roll back" of non-transactional tables by removing the table,
    and the actual rollback might generate events that should not be
    written to the binary log.

3572 3573 3574 3575
  */
  tmp_disable_binlog(thd);
  select_insert::send_error(errcode, err);
  reenable_binlog(thd);
3576 3577

  DBUG_VOID_RETURN;
unknown's avatar
unknown committed
3578 3579
}

unknown's avatar
unknown committed
3580

unknown's avatar
unknown committed
3581 3582 3583 3584 3585 3586 3587
bool select_create::send_eof()
{
  bool tmp=select_insert::send_eof();
  if (tmp)
    abort();
  else
  {
3588 3589 3590 3591 3592 3593 3594 3595
    /*
      Do an implicit commit at end of statement for non-temporary
      tables.  This can fail, but we should unlock the table
      nevertheless.
    */
    if (!table->s->tmp_table)
      ha_commit(thd);               // Can fail, but we proceed anyway

unknown's avatar
unknown committed
3596
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
3597
    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
unknown's avatar
unknown committed
3598
    if (thd->extra_lock)
3599
    {
unknown's avatar
unknown committed
3600 3601
      mysql_unlock_tables(thd, thd->extra_lock);
      thd->extra_lock=0;
3602
    }
unknown's avatar
unknown committed
3603 3604 3605 3606
  }
  return tmp;
}

unknown's avatar
unknown committed
3607

unknown's avatar
unknown committed
3608 3609
void select_create::abort()
{
3610 3611
  DBUG_ENTER("select_create::abort");

3612 3613 3614 3615 3616 3617 3618 3619
  /*
   Disable binlog, because we "roll back" partial inserts in ::abort
   by removing the table, even for non-transactional tables.
  */
  tmp_disable_binlog(thd);
  select_insert::abort();
  reenable_binlog(thd);

3620 3621 3622 3623 3624 3625 3626 3627 3628 3629 3630 3631 3632 3633 3634 3635
  /*
    We roll back the statement, including truncating the transaction
    cache of the binary log, if the statement failed.

    We roll back the statement prior to deleting the table and prior
    to releasing the lock on the table, since there might be potential
    for failure if the rollback is executed after the drop or after
    unlocking the table.

    We also roll back the statement regardless of whether the creation
    of the table succeeded or not, since we need to reset the binary
    log state.
  */
  if (thd->current_stmt_binlog_row_based)
    ha_rollback_stmt(thd);

3636
  if (thd->extra_lock)
unknown's avatar
unknown committed
3637
  {
3638 3639
    mysql_unlock_tables(thd, thd->extra_lock);
    thd->extra_lock=0;
unknown's avatar
unknown committed
3640
  }
3641

unknown's avatar
unknown committed
3642 3643
  if (table)
  {
unknown's avatar
unknown committed
3644
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
3645
    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
unknown's avatar
unknown committed
3646 3647
    if (!create_info->table_existed)
      drop_open_table(thd, table, create_table->db, create_table->table_name);
unknown's avatar
unknown committed
3648
    table=0;                                    // Safety
unknown's avatar
unknown committed
3649
  }
3650
  DBUG_VOID_RETURN;
unknown's avatar
unknown committed
3651 3652 3653 3654
}


/*****************************************************************************
unknown's avatar
unknown committed
3655
  Instansiate templates
unknown's avatar
unknown committed
3656 3657
*****************************************************************************/

3658
#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
unknown's avatar
unknown committed
3659
template class List_iterator_fast<List_item>;
3660
#ifndef EMBEDDED_LIBRARY
3661 3662
template class I_List<Delayed_insert>;
template class I_List_iterator<Delayed_insert>;
unknown's avatar
unknown committed
3663
template class I_List<delayed_row>;
3664
#endif /* EMBEDDED_LIBRARY */
3665
#endif /* HAVE_EXPLICIT_TEMPLATE_INSTANTIATION */