← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_load.cc

  • Committer: Brian Aker
  • Date: 2009-04-27 14:36:40 UTC
  • Revision ID: brian@gaz-20090427143640-f6zjmtt9vm55qgm2
Patch on show processlist from  davi@apache.org

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_load.cc
17
17
/* Copy data from a textfile to table */
18
18
 
19
19
#include <drizzled/server_includes.h>
20
 
#include "sql_repl.h"
21
 
#include <drizzled/drizzled_error_messages.h>
22
 
 
 
20
#include <drizzled/sql_load.h>
 
21
#include <drizzled/error.h>
 
22
#include <drizzled/data_home.h>
 
23
#include <drizzled/session.h>
 
24
#include <drizzled/sql_base.h>
 
25
#include <drizzled/field/timestamp.h>
23
26
 
24
27
class READ_INFO {
25
28
  File  file;
34
37
  bool  found_end_of_line,start_of_line,eof;
35
38
  bool  need_end_io_cache;
36
39
  IO_CACHE cache;
37
 
  NET *io_net;
38
40
 
39
41
public:
40
42
  bool error,line_cuted,found_null,enclosed;
44
46
 
45
47
  READ_INFO(File file,uint32_t tot_length, const CHARSET_INFO * const cs,
46
48
            String &field_term,String &line_start,String &line_term,
47
 
            String &enclosed,int escape,bool get_it_from_net, bool is_fifo);
 
49
            String &enclosed,int escape, bool is_fifo);
48
50
  ~READ_INFO();
49
51
  int read_field();
50
52
  int read_fixed_length(void);
66
68
  /*
67
69
    Either this method, or we need to make cache public
68
70
    Arg must be set from mysql_load() since constructor does not see
69
 
    either the table or THD value
 
71
    either the table or Session value
70
72
  */
71
73
  void set_io_cache_arg(void* arg) { cache.arg = arg; }
72
74
};
73
75
 
74
 
static int read_fixed_length(THD *thd, COPY_INFO &info, TableList *table_list,
 
76
static int read_fixed_length(Session *session, COPY_INFO &info, TableList *table_list,
75
77
                             List<Item> &fields_vars, List<Item> &set_fields,
76
78
                             List<Item> &set_values, READ_INFO &read_info,
77
79
                             uint32_t skip_lines,
78
80
                             bool ignore_check_option_errors);
79
 
static int read_sep_field(THD *thd, COPY_INFO &info, TableList *table_list,
 
81
static int read_sep_field(Session *session, COPY_INFO &info, TableList *table_list,
80
82
                          List<Item> &fields_vars, List<Item> &set_fields,
81
83
                          List<Item> &set_values, READ_INFO &read_info,
82
84
                          String &enclosed, uint32_t skip_lines,
83
85
                          bool ignore_check_option_errors);
84
86
 
85
 
static bool write_execute_load_query_log_event(THD *thd,
86
 
                                               bool duplicates, bool ignore,
87
 
                                               bool transactional_table,
88
 
                                               THD::killed_state killed_status);
89
87
 
90
88
/*
91
89
  Execute LOAD DATA query
92
90
 
93
91
  SYNOPSYS
94
92
    mysql_load()
95
 
      thd - current thread
96
 
      ex  - sql_exchange object representing source file and its parsing rules
 
93
      session - current thread
 
94
      ex  - file_exchange object representing source file and its parsing rules
97
95
      table_list  - list of tables to which we are loading data
98
96
      fields_vars - list of fields and variables to which we read
99
97
                    data from file
102
100
      handle_duplicates - indicates whenever we should emit error or
103
101
                          replace row if we will meet duplicates.
104
102
      ignore -          - indicates whenever we should ignore duplicates
105
 
      read_file_from_client - is this LOAD DATA LOCAL ?
106
103
 
107
104
  RETURN VALUES
108
105
    true - error / false - success
109
106
*/
110
107
 
111
 
int mysql_load(THD *thd,sql_exchange *ex,TableList *table_list,
 
108
int mysql_load(Session *session,file_exchange *ex,TableList *table_list,
112
109
                List<Item> &fields_vars, List<Item> &set_fields,
113
110
                List<Item> &set_values,
114
 
                enum enum_duplicates handle_duplicates, bool ignore,
115
 
                bool read_file_from_client)
 
111
                enum enum_duplicates handle_duplicates, bool ignore)
116
112
{
117
113
  char name[FN_REFLEN];
118
114
  File file;
121
117
  String *field_term=ex->field_term,*escaped=ex->escaped;
122
118
  String *enclosed=ex->enclosed;
123
119
  bool is_fifo=0;
124
 
  LOAD_FILE_INFO lf_info;
125
 
  char *db = table_list->db;                    // This is never null
 
120
  char *db= table_list->db;                     // This is never null
 
121
  assert(db);
126
122
  /*
127
123
    If path for file is not defined, we will use the current database.
128
124
    If this is not set, we will use the directory where the table to be
129
125
    loaded is located
130
126
  */
131
 
  char *tdb= thd->db ? thd->db : db;            // Result is never null
 
127
  char *tdb= session->db ? session->db : db;            // Result is never null
 
128
  assert(tdb);
132
129
  uint32_t skip_lines= ex->skip_lines;
133
130
  bool transactional_table;
134
 
  THD::killed_state killed_status= THD::NOT_KILLED;
 
131
  Session::killed_state killed_status= Session::NOT_KILLED;
135
132
 
136
 
  if (escaped->length() > 1 || enclosed->length() > 1)
 
133
  /* Escape and enclosed character may be a utf8 4-byte character */
 
134
  if (escaped->length() > 4 || enclosed->length() > 4)
137
135
  {
138
 
    my_message(ER_WRONG_FIELD_TERMINATORS,ER(ER_WRONG_FIELD_TERMINATORS),
139
 
               MYF(0));
 
136
    my_error(ER_WRONG_FIELD_TERMINATORS,MYF(0),enclosed->c_ptr(), enclosed->length());
140
137
    return(true);
141
138
  }
142
 
  if (open_and_lock_tables(thd, table_list))
 
139
  if (open_and_lock_tables(session, table_list))
143
140
    return(true);
144
 
  if (setup_tables_and_check_access(thd, &thd->lex->select_lex.context,
145
 
                                    &thd->lex->select_lex.top_join_list,
 
141
  if (setup_tables_and_check_access(session, &session->lex->select_lex.context,
 
142
                                    &session->lex->select_lex.top_join_list,
146
143
                                    table_list,
147
 
                                    &thd->lex->select_lex.leaf_tables, true))
 
144
                                    &session->lex->select_lex.leaf_tables, true))
148
145
     return(-1);
149
146
 
150
147
  /*
155
152
    table is marked to be 'used for insert' in which case we should never
156
153
    mark this table as 'const table' (ie, one that has only one row).
157
154
  */
158
 
  if (unique_table(thd, table_list, table_list->next_global, 0))
 
155
  if (unique_table(session, table_list, table_list->next_global, 0))
159
156
  {
160
157
    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
161
158
    return(true);
169
166
    Field **field;
170
167
    for (field=table->field; *field ; field++)
171
168
      fields_vars.push_back(new Item_field(*field));
172
 
    bitmap_set_all(table->write_set);
 
169
    table->write_set->set();
173
170
    table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
174
171
    /*
175
172
      Let us also prepare SET clause, altough it is probably empty
176
173
      in this case.
177
174
    */
178
 
    if (setup_fields(thd, 0, set_fields, MARK_COLUMNS_WRITE, 0, 0) ||
179
 
        setup_fields(thd, 0, set_values, MARK_COLUMNS_READ, 0, 0))
 
175
    if (setup_fields(session, 0, set_fields, MARK_COLUMNS_WRITE, 0, 0) ||
 
176
        setup_fields(session, 0, set_values, MARK_COLUMNS_READ, 0, 0))
180
177
      return(true);
181
178
  }
182
179
  else
183
180
  {                                             // Part field list
184
181
    /* TODO: use this conds for 'WITH CHECK OPTIONS' */
185
 
    if (setup_fields(thd, 0, fields_vars, MARK_COLUMNS_WRITE, 0, 0) ||
186
 
        setup_fields(thd, 0, set_fields, MARK_COLUMNS_WRITE, 0, 0) ||
187
 
        check_that_all_fields_are_given_values(thd, table, table_list))
 
182
    if (setup_fields(session, 0, fields_vars, MARK_COLUMNS_WRITE, 0, 0) ||
 
183
        setup_fields(session, 0, set_fields, MARK_COLUMNS_WRITE, 0, 0) ||
 
184
        check_that_all_fields_are_given_values(session, table, table_list))
188
185
      return(true);
189
186
    /*
190
187
      Check whenever TIMESTAMP field with auto-set feature specified
192
189
    */
193
190
    if (table->timestamp_field)
194
191
    {
195
 
      if (bitmap_is_set(table->write_set,
196
 
                        table->timestamp_field->field_index))
 
192
      if (table->write_set->test(table->timestamp_field->field_index))
197
193
        table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
198
194
      else
199
195
      {
200
 
        bitmap_set_bit(table->write_set,
201
 
                       table->timestamp_field->field_index);
 
196
        table->write_set->set(table->timestamp_field->field_index);
202
197
      }
203
198
    }
204
199
    /* Fix the expressions in SET clause */
205
 
    if (setup_fields(thd, 0, set_values, MARK_COLUMNS_READ, 0, 0))
 
200
    if (setup_fields(session, 0, set_values, MARK_COLUMNS_READ, 0, 0))
206
201
      return(true);
207
202
  }
208
203
 
243
238
    return(true);
244
239
  }
245
240
 
246
 
  /* We can't give an error in the middle when using LOCAL files */
247
 
  if (read_file_from_client && handle_duplicates == DUP_ERROR)
248
 
    ignore= 1;
249
 
 
250
 
  if (read_file_from_client)
251
 
  {
252
 
    (void)net_request_file(&thd->net,ex->file_name);
253
 
    file = -1;
254
 
  }
255
 
  else
256
241
  {
257
242
#ifdef DONT_ALLOW_FULL_LOAD_DATA_PATHS
258
243
    ex->file_name+=dirname_length(ex->file_name);
259
244
#endif
260
245
    if (!dirname_length(ex->file_name))
261
246
    {
262
 
      strxnmov(name, FN_REFLEN-1, mysql_real_data_home, tdb, NULL);
 
247
      strcpy(name, drizzle_real_data_home);
 
248
      strncat(name, tdb, FN_REFLEN-strlen(drizzle_real_data_home)-1);
263
249
      (void) fn_format(name, ex->file_name, name, "",
264
250
                       MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
265
251
    }
266
252
    else
267
253
    {
268
 
      (void) fn_format(name, ex->file_name, mysql_real_data_home, "",
 
254
      (void) fn_format(name, ex->file_name, drizzle_real_data_home, "",
269
255
                       MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
270
256
 
271
257
      if (opt_secure_file_priv &&
278
264
 
279
265
      struct stat stat_info;
280
266
      if (stat(name,&stat_info))
 
267
      {
 
268
        my_error(ER_FILE_NOT_FOUND, MYF(0), name, errno);
281
269
        return(true);
 
270
      }
282
271
 
283
272
      // if we are not in slave thread, the file must be:
284
 
      if (!thd->slave_thread &&
285
 
          !((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
286
 
            (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
287
 
            ((stat_info.st_mode & S_IFREG) == S_IFREG ||
288
 
             (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
 
273
      if (!((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
 
274
            (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
 
275
            ((stat_info.st_mode & S_IFREG) == S_IFREG ||
 
276
             (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
289
277
      {
290
278
        my_error(ER_TEXTFILE_NOT_READABLE, MYF(0), name);
291
279
        return(true);
294
282
        is_fifo = 1;
295
283
    }
296
284
    if ((file=my_open(name,O_RDONLY,MYF(MY_WME))) < 0)
 
285
    {
 
286
      my_error(ER_CANT_OPEN_FILE, MYF(0), name, my_errno);
297
287
      return(true);
 
288
    }
298
289
  }
299
290
 
300
291
  COPY_INFO info;
304
295
  info.escape_char=escaped->length() ? (*escaped)[0] : INT_MAX;
305
296
 
306
297
  READ_INFO read_info(file,tot_length,
307
 
                      ex->cs ? ex->cs : thd->variables.collation_database,
 
298
                      ex->cs ? ex->cs : session->variables.collation_database,
308
299
                      *field_term,*ex->line_start, *ex->line_term, *enclosed,
309
 
                      info.escape_char, read_file_from_client, is_fifo);
 
300
                      info.escape_char, is_fifo);
310
301
  if (read_info.error)
311
302
  {
312
303
    if  (file >= 0)
314
305
    return(true);                               // Can't allocate buffers
315
306
  }
316
307
 
317
 
  if (mysql_bin_log.is_open())
318
 
  {
319
 
    lf_info.thd = thd;
320
 
    lf_info.wrote_create_file = 0;
321
 
    lf_info.last_pos_in_file = HA_POS_ERROR;
322
 
    lf_info.log_delayed= transactional_table;
323
 
    read_info.set_io_cache_arg((void*) &lf_info);
324
 
  }
325
 
 
326
 
  thd->count_cuted_fields= CHECK_FIELD_WARN;            /* calc cuted fields */
327
 
  thd->cuted_fields=0L;
 
308
  /*
 
309
   * Per the SQL standard, inserting NULL into a NOT NULL
 
310
   * field requires an error to be thrown.
 
311
   *
 
312
   * @NOTE
 
313
   *
 
314
   * NULL check and handling occurs in field_conv.cc
 
315
   */
 
316
  session->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
 
317
  session->cuted_fields=0L;
328
318
  /* Skip lines if there is a line terminator */
329
319
  if (ex->line_term->length())
330
320
  {
349
339
    table->file->ha_start_bulk_insert((ha_rows) 0);
350
340
    table->copy_blobs=1;
351
341
 
352
 
    thd->abort_on_warning= true;
 
342
    session->abort_on_warning= true;
353
343
 
354
344
    if (!field_term->length() && !enclosed->length())
355
 
      error= read_fixed_length(thd, info, table_list, fields_vars,
 
345
      error= read_fixed_length(session, info, table_list, fields_vars,
356
346
                               set_fields, set_values, read_info,
357
347
                               skip_lines, ignore);
358
348
    else
359
 
      error= read_sep_field(thd, info, table_list, fields_vars,
 
349
      error= read_sep_field(session, info, table_list, fields_vars,
360
350
                            set_fields, set_values, read_info,
361
351
                            *enclosed, skip_lines, ignore);
362
352
    if (table->file->ha_end_bulk_insert() && !error)
372
362
    my_close(file,MYF(0));
373
363
  free_blobs(table);                            /* if pack_blob was used */
374
364
  table->copy_blobs=0;
375
 
  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
376
 
  /* 
 
365
  session->count_cuted_fields= CHECK_FIELD_IGNORE;
 
366
  /*
377
367
     simulated killing in the middle of per-row loop
378
368
     must be effective for binlogging
379
369
  */
380
 
  killed_status= (error == 0)? THD::NOT_KILLED : thd->killed;
 
370
  killed_status= (error == 0)? Session::NOT_KILLED : session->killed;
381
371
  if (error)
382
372
  {
383
 
    if (read_file_from_client)
384
 
      while (!read_info.next_line())
385
 
        ;
386
 
 
387
 
    if (mysql_bin_log.is_open())
388
 
    {
389
 
      {
390
 
        /*
391
 
          Make sure last block (the one which caused the error) gets
392
 
          logged.  This is needed because otherwise after write of (to
393
 
          the binlog, not to read_info (which is a cache))
394
 
          Delete_file_log_event the bad block will remain in read_info
395
 
          (because pre_read is not called at the end of the last
396
 
          block; remember pre_read is called whenever a new block is
397
 
          read from disk).  At the end of mysql_load(), the destructor
398
 
          of read_info will call end_io_cache() which will flush
399
 
          read_info, so we will finally have this in the binlog:
400
 
 
401
 
          Append_block # The last successfull block
402
 
          Delete_file
403
 
          Append_block # The failing block
404
 
          which is nonsense.
405
 
          Or could also be (for a small file)
406
 
          Create_file  # The failing block
407
 
          which is nonsense (Delete_file is not written in this case, because:
408
 
          Create_file has not been written, so Delete_file is not written, then
409
 
          when read_info is destroyed end_io_cache() is called which writes
410
 
          Create_file.
411
 
        */
412
 
        read_info.end_io_cache();
413
 
        /* If the file was not empty, wrote_create_file is true */
414
 
        if (lf_info.wrote_create_file)
415
 
        {
416
 
          if (thd->transaction.stmt.modified_non_trans_table)
417
 
            write_execute_load_query_log_event(thd, handle_duplicates,
418
 
                                               ignore, transactional_table,
419
 
                                               killed_status);
420
 
          else
421
 
          {
422
 
            Delete_file_log_event d(thd, db, transactional_table);
423
 
            d.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
424
 
            mysql_bin_log.write(&d);
425
 
          }
426
 
        }
427
 
      }
428
 
    }
429
373
    error= -1;                          // Error on read
430
374
    goto err;
431
375
  }
432
376
  sprintf(name, ER(ER_LOAD_INFO), (uint32_t) info.records, (uint32_t) info.deleted,
433
 
          (uint32_t) (info.records - info.copied), (uint32_t) thd->cuted_fields);
434
 
 
435
 
  if (thd->transaction.stmt.modified_non_trans_table)
436
 
    thd->transaction.all.modified_non_trans_table= true;
437
 
 
438
 
  if (mysql_bin_log.is_open())
439
 
  {
440
 
    /*
441
 
      We need to do the job that is normally done inside
442
 
      binlog_query() here, which is to ensure that the pending event
443
 
      is written before tables are unlocked and before any other
444
 
      events are written.  We also need to update the table map
445
 
      version for the binary log to mark that table maps are invalid
446
 
      after this point.
447
 
     */
448
 
    if (thd->current_stmt_binlog_row_based)
449
 
      thd->binlog_flush_pending_rows_event(true);
450
 
    else
451
 
    {
452
 
      /*
453
 
        As already explained above, we need to call end_io_cache() or the last
454
 
        block will be logged only after Execute_load_query_log_event (which is
455
 
        wrong), when read_info is destroyed.
456
 
      */
457
 
      read_info.end_io_cache();
458
 
      if (lf_info.wrote_create_file)
459
 
      {
460
 
        write_execute_load_query_log_event(thd, handle_duplicates, ignore,
461
 
                                           transactional_table,killed_status);
462
 
      }
463
 
    }
464
 
  }
 
377
          (uint32_t) (info.records - info.copied), (uint32_t) session->cuted_fields);
 
378
 
 
379
  if (session->transaction.stmt.modified_non_trans_table)
 
380
    session->transaction.all.modified_non_trans_table= true;
465
381
 
466
382
  /* ok to client sent only after binlog write and engine commit */
467
 
  my_ok(thd, info.copied + info.deleted, 0L, name);
 
383
  session->my_ok(info.copied + info.deleted, 0L, name);
468
384
err:
469
385
  assert(transactional_table || !(info.copied || info.deleted) ||
470
 
              thd->transaction.stmt.modified_non_trans_table);
 
386
              session->transaction.stmt.modified_non_trans_table);
471
387
  table->file->ha_release_auto_increment();
472
388
  table->auto_increment_field_not_null= false;
473
 
  thd->abort_on_warning= 0;
 
389
  session->abort_on_warning= 0;
474
390
  return(error);
475
391
}
476
392
 
477
393
 
478
 
/* Not a very useful function; just to avoid duplication of code */
479
 
static bool write_execute_load_query_log_event(THD *thd,
480
 
                                               bool duplicates, bool ignore,
481
 
                                               bool transactional_table,
482
 
                                               THD::killed_state killed_err_arg)
483
 
{
484
 
  Execute_load_query_log_event
485
 
    e(thd, thd->query, thd->query_length,
486
 
      (char*)thd->lex->fname_start - (char*)thd->query,
487
 
      (char*)thd->lex->fname_end - (char*)thd->query,
488
 
      (duplicates == DUP_REPLACE) ? LOAD_DUP_REPLACE :
489
 
      (ignore ? LOAD_DUP_IGNORE : LOAD_DUP_ERROR),
490
 
      transactional_table, false, killed_err_arg);
491
 
  e.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
492
 
  return mysql_bin_log.write(&e);
493
 
}
494
 
 
495
 
 
496
394
/****************************************************************************
497
395
** Read of rows of fixed size + optional garage + optonal newline
498
396
****************************************************************************/
499
397
 
500
398
static int
501
 
read_fixed_length(THD *thd, COPY_INFO &info, TableList *table_list,
 
399
read_fixed_length(Session *session, COPY_INFO &info, TableList *table_list,
502
400
                  List<Item> &fields_vars, List<Item> &set_fields,
503
401
                  List<Item> &set_values, READ_INFO &read_info,
504
402
                  uint32_t skip_lines, bool ignore_check_option_errors)
510
408
  bool err;
511
409
 
512
410
  id= 0;
513
 
 
 
411
 
514
412
  while (!read_info.read_fixed_length())
515
413
  {
516
 
    if (thd->killed)
 
414
    if (session->killed)
517
415
    {
518
 
      thd->send_kill_message();
 
416
      session->send_kill_message();
519
417
      return(1);
520
418
    }
521
419
    if (skip_lines)
542
440
    */
543
441
    while ((sql_field= (Item_field*) it++))
544
442
    {
545
 
      Field *field= sql_field->field;                  
 
443
      Field *field= sql_field->field;
546
444
      if (field == table->next_number_field)
547
445
        table->auto_increment_field_not_null= true;
548
446
      /*
554
452
 
555
453
      if (pos == read_info.row_end)
556
454
      {
557
 
        thd->cuted_fields++;                    /* Not enough fields */
558
 
        push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN, 
559
 
                            ER_WARN_TOO_FEW_RECORDS, 
560
 
                            ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
 
455
        session->cuted_fields++;                        /* Not enough fields */
 
456
        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
457
                            ER_WARN_TOO_FEW_RECORDS,
 
458
                            ER(ER_WARN_TOO_FEW_RECORDS), session->row_count);
561
459
        if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
562
460
            ((Field_timestamp*) field)->set_time();
563
461
      }
565
463
      {
566
464
        uint32_t length;
567
465
        unsigned char save_chr;
568
 
        if ((length=(uint) (read_info.row_end-pos)) >
 
466
        if ((length=(uint32_t) (read_info.row_end-pos)) >
569
467
            field->field_length)
570
468
          length=field->field_length;
571
469
        save_chr=pos[length]; pos[length]='\0'; // Safeguard aganst malloc
577
475
    }
578
476
    if (pos != read_info.row_end)
579
477
    {
580
 
      thd->cuted_fields++;                      /* To long row */
581
 
      push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN, 
582
 
                          ER_WARN_TOO_MANY_RECORDS, 
583
 
                          ER(ER_WARN_TOO_MANY_RECORDS), thd->row_count); 
 
478
      session->cuted_fields++;                  /* To long row */
 
479
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
480
                          ER_WARN_TOO_MANY_RECORDS,
 
481
                          ER(ER_WARN_TOO_MANY_RECORDS), session->row_count);
584
482
    }
585
483
 
586
 
    if (thd->killed ||
587
 
        fill_record(thd, set_fields, set_values,
 
484
    if (session->killed ||
 
485
        fill_record(session, set_fields, set_values,
588
486
                    ignore_check_option_errors))
589
487
      return(1);
590
488
 
591
 
    err= write_record(thd, table, &info);
 
489
    err= write_record(session, table, &info);
592
490
    table->auto_increment_field_not_null= false;
593
491
    if (err)
594
492
      return(1);
595
 
   
 
493
 
596
494
    /*
597
495
      We don't need to reset auto-increment field since we are restoring
598
496
      its default value at the beginning of each loop iteration.
601
499
      break;
602
500
    if (read_info.line_cuted)
603
501
    {
604
 
      thd->cuted_fields++;                      /* To long row */
605
 
      push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN, 
606
 
                          ER_WARN_TOO_MANY_RECORDS, 
607
 
                          ER(ER_WARN_TOO_MANY_RECORDS), thd->row_count); 
 
502
      session->cuted_fields++;                  /* To long row */
 
503
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
504
                          ER_WARN_TOO_MANY_RECORDS,
 
505
                          ER(ER_WARN_TOO_MANY_RECORDS), session->row_count);
608
506
    }
609
 
    thd->row_count++;
 
507
    session->row_count++;
610
508
  }
611
509
  return(test(read_info.error));
612
510
}
614
512
 
615
513
 
616
514
static int
617
 
read_sep_field(THD *thd, COPY_INFO &info, TableList *table_list,
 
515
read_sep_field(Session *session, COPY_INFO &info, TableList *table_list,
618
516
               List<Item> &fields_vars, List<Item> &set_fields,
619
517
               List<Item> &set_values, READ_INFO &read_info,
620
518
               String &enclosed, uint32_t skip_lines,
632
530
 
633
531
  for (;;it.rewind())
634
532
  {
635
 
    if (thd->killed)
 
533
    if (session->killed)
636
534
    {
637
 
      thd->send_kill_message();
 
535
      session->send_kill_message();
638
536
      return(1);
639
537
    }
640
538
 
654
552
        continue;
655
553
 
656
554
      pos=read_info.row_start;
657
 
      length=(uint) (read_info.row_end-pos);
 
555
      length=(uint32_t) (read_info.row_end-pos);
658
556
 
659
557
      real_item= item->real_item();
660
558
 
668
566
          if (field->reset())
669
567
          {
670
568
            my_error(ER_WARN_NULL_TO_NOTNULL, MYF(0), field->field_name,
671
 
                     thd->row_count);
 
569
                     session->row_count);
672
570
            return(1);
673
571
          }
674
572
          field->set_null();
736
634
          if (field->reset())
737
635
          {
738
636
            my_error(ER_WARN_NULL_TO_NOTNULL, MYF(0),field->field_name,
739
 
                     thd->row_count);
 
637
                     session->row_count);
740
638
            return(1);
741
639
          }
742
640
          if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
744
642
          /*
745
643
            QQ: We probably should not throw warning for each field.
746
644
            But how about intention to always have the same number
747
 
            of warnings in THD::cuted_fields (and get rid of cuted_fields
 
645
            of warnings in Session::cuted_fields (and get rid of cuted_fields
748
646
            in the end ?)
749
647
          */
750
 
          thd->cuted_fields++;
751
 
          push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
648
          session->cuted_fields++;
 
649
          push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
752
650
                              ER_WARN_TOO_FEW_RECORDS,
753
 
                              ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
 
651
                              ER(ER_WARN_TOO_FEW_RECORDS), session->row_count);
754
652
        }
755
653
        else if (item->type() == Item::STRING_ITEM)
756
654
        {
765
663
      }
766
664
    }
767
665
 
768
 
    if (thd->killed ||
769
 
        fill_record(thd, set_fields, set_values,
 
666
    if (session->killed ||
 
667
        fill_record(session, set_fields, set_values,
770
668
                    ignore_check_option_errors))
771
669
      return(1);
772
670
 
773
 
    err= write_record(thd, table, &info);
 
671
    err= write_record(session, table, &info);
774
672
    table->auto_increment_field_not_null= false;
775
673
    if (err)
776
674
      return(1);
782
680
      break;
783
681
    if (read_info.line_cuted)
784
682
    {
785
 
      thd->cuted_fields++;                      /* To long row */
786
 
      push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN, 
787
 
                          ER_WARN_TOO_MANY_RECORDS, ER(ER_WARN_TOO_MANY_RECORDS), 
788
 
                          thd->row_count);   
789
 
      if (thd->killed)
 
683
      session->cuted_fields++;                  /* To long row */
 
684
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
685
                          ER_WARN_TOO_MANY_RECORDS, ER(ER_WARN_TOO_MANY_RECORDS),
 
686
                          session->row_count);
 
687
      if (session->killed)
790
688
        return(1);
791
689
    }
792
 
    thd->row_count++;
 
690
    session->row_count++;
793
691
  }
794
692
  return(test(read_info.error));
795
693
}
824
722
 
825
723
READ_INFO::READ_INFO(File file_par, uint32_t tot_length, const CHARSET_INFO * const cs,
826
724
                     String &field_term, String &line_start, String &line_term,
827
 
                     String &enclosed_par, int escape, bool get_it_from_net,
828
 
                     bool is_fifo)
 
725
                     String &enclosed_par, int escape, bool is_fifo)
829
726
  :file(file_par),escape_char(escape)
830
727
{
831
728
  read_charset= cs;
864
761
  set_if_bigger(length,line_start.length());
865
762
  stack=stack_pos=(int*) sql_alloc(sizeof(int)*length);
866
763
 
867
 
  if (!(buffer=(unsigned char*) my_malloc(buff_length+1,MYF(0))))
 
764
  if (!(buffer=(unsigned char*) malloc(buff_length+1)))
868
765
    error=1; /* purecov: inspected */
869
766
  else
870
767
  {
871
768
    end_of_buff=buffer+buff_length;
872
 
    if (init_io_cache(&cache,(get_it_from_net) ? -1 : file, 0,
873
 
                      (get_it_from_net) ? READ_NET :
 
769
    if (init_io_cache(&cache,(false) ? -1 : file, 0,
 
770
                      (false) ? READ_NET :
874
771
                      (is_fifo ? READ_FIFO : READ_CACHE),0L,1,
875
772
                      MYF(MY_WME)))
876
773
    {
885
782
        manual assignment
886
783
      */
887
784
      need_end_io_cache = 1;
888
 
 
889
 
      if (get_it_from_net)
890
 
        cache.read_function = _my_b_net_read;
891
 
 
892
 
      if (mysql_bin_log.is_open())
893
 
        cache.pre_read = cache.pre_close =
894
 
          (IO_CACHE_CALLBACK) log_loaded_block;
895
785
    }
896
786
  }
897
787
}
1082
972
    /*
1083
973
    ** We come here if buffer is too small. Enlarge it and continue
1084
974
    */
1085
 
    if (!(new_buffer=(unsigned char*) my_realloc((char*) buffer,buff_length+1+IO_SIZE,
1086
 
                                        MYF(MY_WME))))
 
975
    if (!(new_buffer=(unsigned char*) realloc(buffer, buff_length+1+IO_SIZE)))
1087
976
      return (error=1);
1088
977
    to=new_buffer + (to-buffer);
1089
978
    buffer=new_buffer;