← 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: 2010-10-19 04:37:10 UTC
  • mto: (1861.1.2 build) (1864.2.1 merge)
  • mto: This revision was merged to the branch mainline in revision 1862.
  • Revision ID: brian@tangent.org-20101019043710-anbhydvze74xvesn
A few cleanups.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_load.cc
16
16
 
17
17
/* Copy data from a textfile to table */
18
18
 
19
 
#include <config.h>
20
 
 
 
19
#include "config.h"
21
20
#include <drizzled/sql_load.h>
22
21
#include <drizzled/error.h>
23
22
#include <drizzled/data_home.h>
24
23
#include <drizzled/session.h>
25
24
#include <drizzled/sql_base.h>
26
 
#include <drizzled/field/epoch.h>
27
 
#include <drizzled/internal/my_sys.h>
28
 
#include <drizzled/internal/iocache.h>
29
 
#include <drizzled/plugin/storage_engine.h>
 
25
#include <drizzled/field/timestamp.h>
 
26
#include "drizzled/internal/my_sys.h"
 
27
#include "drizzled/internal/iocache.h"
 
28
#include <drizzled/db.h>
30
29
 
31
30
#include <sys/stat.h>
32
31
#include <fcntl.h>
76
75
  */
77
76
  void end_io_cache()
78
77
  {
79
 
    cache.end_io_cache();
 
78
    internal::end_io_cache(&cache);
80
79
    need_end_io_cache = 0;
81
80
  }
82
81
 
83
82
  /*
84
83
    Either this method, or we need to make cache public
85
 
    Arg must be set from load() since constructor does not see
 
84
    Arg must be set from mysql_load() since constructor does not see
86
85
    either the table or Session value
87
86
  */
88
87
  void set_io_cache_arg(void* arg) { cache.arg = arg; }
104
103
  Execute LOAD DATA query
105
104
 
106
105
  SYNOPSYS
107
 
    load()
 
106
    mysql_load()
108
107
      session - current thread
109
108
      ex  - file_exchange object representing source cursor and its parsing rules
110
109
      table_list  - list of tables to which we are loading data
120
119
    true - error / false - success
121
120
*/
122
121
 
123
 
int load(Session *session,file_exchange *ex,TableList *table_list,
 
122
int mysql_load(Session *session,file_exchange *ex,TableList *table_list,
124
123
                List<Item> &fields_vars, List<Item> &set_fields,
125
124
                List<Item> &set_values,
126
125
                enum enum_duplicates handle_duplicates, bool ignore)
131
130
  String *field_term=ex->field_term,*escaped=ex->escaped;
132
131
  String *enclosed=ex->enclosed;
133
132
  bool is_fifo=0;
134
 
 
135
 
  assert(table_list->getSchemaName()); // This should never be null
136
 
 
 
133
  char *db= table_list->db;                     // This is never null
 
134
  assert(db);
137
135
  /*
138
136
    If path for cursor is not defined, we will use the current database.
139
137
    If this is not set, we will use the directory where the table to be
140
138
    loaded is located
141
139
  */
142
 
  util::string::const_shared_ptr schema(session->schema());
143
 
  const char *tdb= (schema and not schema->empty()) ? schema->c_str() : table_list->getSchemaName(); // Result should never be null
 
140
  const char *tdb= session->db.empty() ? db  : session->db.c_str();             // Result is never null
144
141
  assert(tdb);
145
142
  uint32_t skip_lines= ex->skip_lines;
146
143
  bool transactional_table;
147
 
  Session::killed_state_t killed_status= Session::NOT_KILLED;
 
144
  Session::killed_state killed_status= Session::NOT_KILLED;
148
145
 
149
146
  /* Escape and enclosed character may be a utf8 4-byte character */
150
147
  if (escaped->length() > 4 || enclosed->length() > 4)
156
153
  if (session->openTablesLock(table_list))
157
154
    return(true);
158
155
 
159
 
  if (setup_tables_and_check_access(session, &session->getLex()->select_lex.context,
160
 
                                    &session->getLex()->select_lex.top_join_list,
 
156
  if (setup_tables_and_check_access(session, &session->lex->select_lex.context,
 
157
                                    &session->lex->select_lex.top_join_list,
161
158
                                    table_list,
162
 
                                    &session->getLex()->select_lex.leaf_tables, true))
 
159
                                    &session->lex->select_lex.leaf_tables, true))
163
160
     return(-1);
164
161
 
165
162
  /*
172
169
  */
173
170
  if (unique_table(table_list, table_list->next_global))
174
171
  {
175
 
    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->getTableName());
 
172
    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
176
173
    return(true);
177
174
  }
178
175
 
207
204
    */
208
205
    if (table->timestamp_field)
209
206
    {
210
 
      if (table->isWriteSet(table->timestamp_field->position()))
211
 
      {
 
207
      if (table->isWriteSet(table->timestamp_field->field_index))
212
208
        table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
213
 
      }
214
209
      else
215
210
      {
216
 
        table->setWriteSet(table->timestamp_field->position());
 
211
        table->setWriteSet(table->timestamp_field->field_index);
217
212
      }
218
213
    }
219
214
    /* Fix the expressions in SET clause */
225
220
 
226
221
  size_t tot_length=0;
227
222
  bool use_blobs= 0, use_vars= 0;
228
 
  List<Item>::iterator it(fields_vars.begin());
 
223
  List_iterator_fast<Item> it(fields_vars);
229
224
  Item *item;
230
225
 
231
226
  while ((item= it++))
320
315
  info.handle_duplicates=handle_duplicates;
321
316
  info.escape_char=escaped->length() ? (*escaped)[0] : INT_MAX;
322
317
 
323
 
  identifier::Schema identifier(*schema);
 
318
  SchemaIdentifier identifier(session->db);
324
319
  READ_INFO read_info(file, tot_length,
325
320
                      ex->cs ? ex->cs : plugin::StorageEngine::getSchemaCollation(identifier),
326
321
                      *field_term, *ex->line_start, *ex->line_term, *enclosed,
366
361
    table->cursor->ha_start_bulk_insert((ha_rows) 0);
367
362
    table->copy_blobs=1;
368
363
 
369
 
    session->setAbortOnWarning(true);
 
364
    session->abort_on_warning= true;
370
365
 
371
366
    if (!field_term->length() && !enclosed->length())
372
367
      error= read_fixed_length(session, info, table_list, fields_vars,
394
389
     simulated killing in the middle of per-row loop
395
390
     must be effective for binlogging
396
391
  */
397
 
  killed_status= (error == 0)? Session::NOT_KILLED : session->getKilled();
 
392
  killed_status= (error == 0)? Session::NOT_KILLED : session->killed;
398
393
  if (error)
399
394
  {
400
395
    error= -1;                          // Error on read
415
410
              session->transaction.stmt.hasModifiedNonTransData());
416
411
  table->cursor->ha_release_auto_increment();
417
412
  table->auto_increment_field_not_null= false;
418
 
  session->setAbortOnWarning(false);
419
 
 
 
413
  session->abort_on_warning= 0;
420
414
  return(error);
421
415
}
422
416
 
431
425
                  List<Item> &set_values, READ_INFO &read_info,
432
426
                  uint32_t skip_lines, bool ignore_check_option_errors)
433
427
{
434
 
  List<Item>::iterator it(fields_vars.begin());
 
428
  List_iterator_fast<Item> it(fields_vars);
435
429
  Item_field *sql_field;
436
430
  Table *table= table_list->table;
437
431
  uint64_t id;
441
435
 
442
436
  while (!read_info.read_fixed_length())
443
437
  {
444
 
    if (session->getKilled())
 
438
    if (session->killed)
445
439
    {
446
440
      session->send_kill_message();
447
441
      return(1);
457
451
      skip_lines--;
458
452
      continue;
459
453
    }
460
 
    it= fields_vars.begin();
 
454
    it.rewind();
461
455
    unsigned char *pos=read_info.row_start;
462
 
#ifdef HAVE_VALGRIND
 
456
#ifdef HAVE_purify
463
457
    read_info.row_end[0]=0;
464
458
#endif
465
459
 
486
480
        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
487
481
                            ER_WARN_TOO_FEW_RECORDS,
488
482
                            ER(ER_WARN_TOO_FEW_RECORDS), session->row_count);
489
 
 
490
 
        if (not field->maybe_null() and field->is_timestamp())
491
 
            ((field::Epoch::pointer) field)->set_time();
 
483
        if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
484
            ((Field_timestamp*) field)->set_time();
492
485
      }
493
486
      else
494
487
      {
515
508
                          ER(ER_WARN_TOO_MANY_RECORDS), session->row_count);
516
509
    }
517
510
 
518
 
    if (session->getKilled() ||
 
511
    if (session->killed ||
519
512
        fill_record(session, set_fields, set_values,
520
513
                    ignore_check_option_errors))
521
514
      return(1);
552
545
               String &enclosed, uint32_t skip_lines,
553
546
               bool ignore_check_option_errors)
554
547
{
555
 
  List<Item>::iterator it(fields_vars.begin());
 
548
  List_iterator_fast<Item> it(fields_vars);
556
549
  Item *item;
557
550
  Table *table= table_list->table;
558
551
  uint32_t enclosed_length;
562
555
  enclosed_length=enclosed.length();
563
556
  id= 0;
564
557
 
565
 
  for (;;it= fields_vars.begin())
 
558
  for (;;it.rewind())
566
559
  {
567
 
    if (session->getKilled())
 
560
    if (session->killed)
568
561
    {
569
562
      session->send_kill_message();
570
563
      return(1);
604
597
            return(1);
605
598
          }
606
599
          field->set_null();
607
 
          if (not field->maybe_null())
 
600
          if (!field->maybe_null())
608
601
          {
609
 
            if (field->is_timestamp())
610
 
            {
611
 
              ((field::Epoch::pointer) field)->set_time();
612
 
            }
 
602
            if (field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
603
              ((Field_timestamp*) field)->set_time();
613
604
            else if (field != table->next_number_field)
614
 
            {
615
 
              field->set_warning(DRIZZLE_ERROR::WARN_LEVEL_WARN, ER_WARN_NULL_TO_NOTNULL, 1);
616
 
            }
 
605
              field->set_warning(DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
606
                                 ER_WARN_NULL_TO_NOTNULL, 1);
617
607
          }
618
608
        }
619
609
        else if (item->type() == Item::STRING_ITEM)
674
664
                     session->row_count);
675
665
            return(1);
676
666
          }
677
 
          if (not field->maybe_null() and field->is_timestamp())
678
 
              ((field::Epoch::pointer) field)->set_time();
 
667
          if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
668
              ((Field_timestamp*) field)->set_time();
679
669
          /*
680
670
            QQ: We probably should not throw warning for each field.
681
671
            But how about intention to always have the same number
700
690
      }
701
691
    }
702
692
 
703
 
    if (session->getKilled() ||
 
693
    if (session->killed ||
704
694
        fill_record(session, set_fields, set_values,
705
695
                    ignore_check_option_errors))
706
696
      return(1);
721
711
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
722
712
                          ER_WARN_TOO_MANY_RECORDS, ER(ER_WARN_TOO_MANY_RECORDS),
723
713
                          session->row_count);
724
 
      if (session->getKilled())
 
714
      if (session->killed)
725
715
        return(1);
726
716
    }
727
717
    session->row_count++;
804
794
  else
805
795
  {
806
796
    end_of_buff=buffer+buff_length;
807
 
    if (cache.init_io_cache((false) ? -1 : cursor, 0,
808
 
                            (false) ? internal::READ_NET :
809
 
                            (is_fifo ? internal::READ_FIFO : internal::READ_CACHE),0L,1,
810
 
                            MYF(MY_WME)))
 
797
    if (init_io_cache(&cache,(false) ? -1 : cursor, 0,
 
798
                      (false) ? internal::READ_NET :
 
799
                      (is_fifo ? internal::READ_FIFO : internal::READ_CACHE),0L,1,
 
800
                      MYF(MY_WME)))
811
801
    {
812
802
      free((unsigned char*) buffer);
813
803
      error=1;
830
820
  if (!error)
831
821
  {
832
822
    if (need_end_io_cache)
833
 
      cache.end_io_cache();
 
823
      internal::end_io_cache(&cache);
834
824
    free(buffer);
835
825
    error=1;
836
826
  }