← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_load.cc

  • Committer: Padraig O'Sullivan
  • Date: 2009-07-22 23:26:26 UTC
  • mto: (1039.5.43 replication)
  • mto: This revision was merged to the branch mainline in revision 1130.
  • Revision ID: osullivan.padraig@gmail.com-20090722232626-mu4khq7ho6dqcf7q
Created a simple filtered replicator that can filter by schema name or table
name.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_load.cc
11
11
 
12
12
   You should have received a copy of the GNU General Public License
13
13
   along with this program; if not, write to the Free Software
14
 
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
15
15
 
16
16
 
17
17
/* Copy data from a textfile to table */
18
18
 
19
 
#include "config.h"
20
 
 
 
19
#include <drizzled/server_includes.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/db.h>
30
 
#include "drizzled/plugin/storage_engine.h"
 
25
#include <drizzled/field/timestamp.h>
31
26
 
32
 
#include <sys/stat.h>
33
 
#include <fcntl.h>
34
27
#include <algorithm>
35
 
#include <climits>
36
 
#include <boost/filesystem.hpp>
37
28
 
38
 
namespace fs=boost::filesystem;
39
29
using namespace std;
40
 
namespace drizzled
41
 
{
42
30
 
43
31
class READ_INFO {
44
 
  int   cursor;
 
32
  File  file;
45
33
  unsigned char *buffer;                /* Buffer for read text */
46
34
  unsigned char *end_of_buff;           /* Data in bufferts ends here */
47
35
  size_t buff_length;                   /* Length of buffert */
52
40
  int   *stack,*stack_pos;
53
41
  bool  found_end_of_line,start_of_line,eof;
54
42
  bool  need_end_io_cache;
55
 
  internal::IO_CACHE cache;
 
43
  IO_CACHE cache;
56
44
 
57
45
public:
58
46
  bool error,line_cuted,found_null,enclosed;
60
48
        *row_end;                       /* Found row ends here */
61
49
  const CHARSET_INFO *read_charset;
62
50
 
63
 
  READ_INFO(int cursor, size_t tot_length, const CHARSET_INFO * const cs,
 
51
  READ_INFO(File file, size_t tot_length, const CHARSET_INFO * const cs,
64
52
            String &field_term,String &line_start,String &line_term,
65
53
            String &enclosed,int escape, bool is_fifo);
66
54
  ~READ_INFO();
77
65
  */
78
66
  void end_io_cache()
79
67
  {
80
 
    cache.end_io_cache();
 
68
    ::end_io_cache(&cache);
81
69
    need_end_io_cache = 0;
82
70
  }
83
71
 
84
72
  /*
85
73
    Either this method, or we need to make cache public
86
 
    Arg must be set from load() since constructor does not see
 
74
    Arg must be set from mysql_load() since constructor does not see
87
75
    either the table or Session value
88
76
  */
89
77
  void set_io_cache_arg(void* arg) { cache.arg = arg; }
90
78
};
91
79
 
92
 
static int read_fixed_length(Session *session, CopyInfo &info, TableList *table_list,
 
80
static int read_fixed_length(Session *session, COPY_INFO &info, TableList *table_list,
93
81
                             List<Item> &fields_vars, List<Item> &set_fields,
94
82
                             List<Item> &set_values, READ_INFO &read_info,
95
83
                             uint32_t skip_lines,
96
84
                             bool ignore_check_option_errors);
97
 
static int read_sep_field(Session *session, CopyInfo &info, TableList *table_list,
 
85
static int read_sep_field(Session *session, COPY_INFO &info, TableList *table_list,
98
86
                          List<Item> &fields_vars, List<Item> &set_fields,
99
87
                          List<Item> &set_values, READ_INFO &read_info,
100
88
                          String &enclosed, uint32_t skip_lines,
105
93
  Execute LOAD DATA query
106
94
 
107
95
  SYNOPSYS
108
 
    load()
 
96
    mysql_load()
109
97
      session - current thread
110
 
      ex  - file_exchange object representing source cursor and its parsing rules
 
98
      ex  - file_exchange object representing source file and its parsing rules
111
99
      table_list  - list of tables to which we are loading data
112
100
      fields_vars - list of fields and variables to which we read
113
 
                    data from cursor
 
101
                    data from file
114
102
      set_fields  - list of fields mentioned in set clause
115
103
      set_values  - expressions to assign to fields in previous list
116
104
      handle_duplicates - indicates whenever we should emit error or
121
109
    true - error / false - success
122
110
*/
123
111
 
124
 
int load(Session *session,file_exchange *ex,TableList *table_list,
 
112
int mysql_load(Session *session,file_exchange *ex,TableList *table_list,
125
113
                List<Item> &fields_vars, List<Item> &set_fields,
126
114
                List<Item> &set_values,
127
115
                enum enum_duplicates handle_duplicates, bool ignore)
128
116
{
129
 
  int file;
 
117
  char name[FN_REFLEN];
 
118
  File file;
130
119
  Table *table= NULL;
131
120
  int error;
132
121
  String *field_term=ex->field_term,*escaped=ex->escaped;
133
122
  String *enclosed=ex->enclosed;
134
123
  bool is_fifo=0;
135
 
 
136
 
  assert(table_list->getSchemaName()); // This should never be null
137
 
 
 
124
  char *db= table_list->db;                     // This is never null
 
125
  assert(db);
138
126
  /*
139
 
    If path for cursor is not defined, we will use the current database.
 
127
    If path for file is not defined, we will use the current database.
140
128
    If this is not set, we will use the directory where the table to be
141
129
    loaded is located
142
130
  */
143
 
  util::string::const_shared_ptr schema(session->schema());
144
 
  const char *tdb= (schema and not schema->empty()) ? schema->c_str() : table_list->getSchemaName(); // Result should never be null
 
131
  char *tdb= session->db ? session->db : db;            // Result is never null
145
132
  assert(tdb);
146
133
  uint32_t skip_lines= ex->skip_lines;
147
134
  bool transactional_table;
148
 
  Session::killed_state_t killed_status= Session::NOT_KILLED;
 
135
  Session::killed_state killed_status= Session::NOT_KILLED;
149
136
 
150
137
  /* Escape and enclosed character may be a utf8 4-byte character */
151
138
  if (escaped->length() > 4 || enclosed->length() > 4)
153
140
    my_error(ER_WRONG_FIELD_TERMINATORS,MYF(0),enclosed->c_ptr(), enclosed->length());
154
141
    return(true);
155
142
  }
156
 
 
157
 
  if (session->openTablesLock(table_list))
 
143
  if (session->open_and_lock_tables(table_list))
158
144
    return(true);
159
 
 
160
145
  if (setup_tables_and_check_access(session, &session->lex->select_lex.context,
161
146
                                    &session->lex->select_lex.top_join_list,
162
147
                                    table_list,
171
156
    table is marked to be 'used for insert' in which case we should never
172
157
    mark this table as 'const table' (ie, one that has only one row).
173
158
  */
174
 
  if (unique_table(table_list, table_list->next_global))
 
159
  if (unique_table(session, table_list, table_list->next_global, 0))
175
160
  {
176
 
    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->getTableName());
 
161
    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
177
162
    return(true);
178
163
  }
179
164
 
180
165
  table= table_list->table;
181
 
  transactional_table= table->cursor->has_transactions();
 
166
  transactional_table= table->file->has_transactions();
182
167
 
183
168
  if (!fields_vars.elements)
184
169
  {
185
170
    Field **field;
186
 
    for (field= table->getFields(); *field ; field++)
 
171
    for (field=table->field; *field ; field++)
187
172
      fields_vars.push_back(new Item_field(*field));
188
173
    table->setWriteSet();
189
174
    table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
208
193
    */
209
194
    if (table->timestamp_field)
210
195
    {
211
 
      if (table->isWriteSet(table->timestamp_field->position()))
212
 
      {
 
196
      if (table->isWriteSet(table->timestamp_field->field_index))
213
197
        table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
214
 
      }
215
198
      else
216
199
      {
217
 
        table->setWriteSet(table->timestamp_field->position());
 
200
        table->setWriteSet(table->timestamp_field->field_index);
218
201
      }
219
202
    }
220
203
    /* Fix the expressions in SET clause */
259
242
    return(true);
260
243
  }
261
244
 
262
 
  fs::path to_file(ex->file_name);
263
 
  fs::path target_path(fs::system_complete(getDataHomeCatalog()));
264
 
  if (not to_file.has_root_directory())
265
 
  {
266
 
    int count_elements= 0;
267
 
    for (fs::path::iterator iter= to_file.begin();
268
 
         iter != to_file.end();
269
 
         ++iter, ++count_elements)
270
 
    { }
271
 
 
272
 
    if (count_elements == 1)
273
 
    {
274
 
      target_path /= tdb;
275
 
    }
276
 
    target_path /= to_file;
277
 
  }
278
 
  else
279
 
  {
280
 
    target_path= to_file;
281
 
  }
282
 
 
283
 
  if (not secure_file_priv.string().empty())
284
 
  {
285
 
    if (target_path.file_string().substr(0, secure_file_priv.file_string().size()) != secure_file_priv.file_string())
286
 
    {
287
 
      /* Read only allowed from within dir specified by secure_file_priv */
288
 
      my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
 
245
  {
 
246
#ifdef DONT_ALLOW_FULL_LOAD_DATA_PATHS
 
247
    ex->file_name+=dirname_length(ex->file_name);
 
248
#endif
 
249
    if (!dirname_length(ex->file_name))
 
250
    {
 
251
      strcpy(name, drizzle_real_data_home);
 
252
      strncat(name, tdb, FN_REFLEN-strlen(drizzle_real_data_home)-1);
 
253
      (void) fn_format(name, ex->file_name, name, "",
 
254
                       MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
 
255
    }
 
256
    else
 
257
    {
 
258
      (void) fn_format(name, ex->file_name, drizzle_real_data_home, "",
 
259
                       MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
 
260
 
 
261
      if (opt_secure_file_priv &&
 
262
          strncmp(opt_secure_file_priv, name, strlen(opt_secure_file_priv)))
 
263
      {
 
264
        /* Read only allowed from within dir specified by secure_file_priv */
 
265
        my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
 
266
        return(true);
 
267
      }
 
268
 
 
269
      struct stat stat_info;
 
270
      if (stat(name,&stat_info))
 
271
      {
 
272
        my_error(ER_FILE_NOT_FOUND, MYF(0), name, errno);
 
273
        return(true);
 
274
      }
 
275
 
 
276
      // if we are not in slave thread, the file must be:
 
277
      if (!((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
 
278
            (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
 
279
            ((stat_info.st_mode & S_IFREG) == S_IFREG ||
 
280
             (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
 
281
      {
 
282
        my_error(ER_TEXTFILE_NOT_READABLE, MYF(0), name);
 
283
        return(true);
 
284
      }
 
285
      if ((stat_info.st_mode & S_IFIFO) == S_IFIFO)
 
286
        is_fifo = 1;
 
287
    }
 
288
    if ((file=my_open(name,O_RDONLY,MYF(MY_WME))) < 0)
 
289
    {
 
290
      my_error(ER_CANT_OPEN_FILE, MYF(0), name, my_errno);
289
291
      return(true);
290
292
    }
291
293
  }
292
294
 
293
 
  struct stat stat_info;
294
 
  if (stat(target_path.file_string().c_str(), &stat_info))
295
 
  {
296
 
    my_error(ER_FILE_NOT_FOUND, MYF(0), target_path.file_string().c_str(), errno);
297
 
    return(true);
298
 
  }
299
 
 
300
 
  // if we are not in slave thread, the cursor must be:
301
 
  if (!((stat_info.st_mode & S_IROTH) == S_IROTH &&  // readable by others
302
 
        (stat_info.st_mode & S_IFLNK) != S_IFLNK && // and not a symlink
303
 
        ((stat_info.st_mode & S_IFREG) == S_IFREG ||
304
 
         (stat_info.st_mode & S_IFIFO) == S_IFIFO)))
305
 
  {
306
 
    my_error(ER_TEXTFILE_NOT_READABLE, MYF(0), target_path.file_string().c_str());
307
 
    return(true);
308
 
  }
309
 
  if ((stat_info.st_mode & S_IFIFO) == S_IFIFO)
310
 
    is_fifo = 1;
311
 
 
312
 
 
313
 
  if ((file=internal::my_open(target_path.file_string().c_str(), O_RDONLY,MYF(MY_WME))) < 0)
314
 
  {
315
 
    my_error(ER_CANT_OPEN_FILE, MYF(0), target_path.file_string().c_str(), errno);
316
 
    return(true);
317
 
  }
318
 
  CopyInfo info;
 
295
  COPY_INFO info;
319
296
  memset(&info, 0, sizeof(info));
320
297
  info.ignore= ignore;
321
298
  info.handle_duplicates=handle_duplicates;
322
299
  info.escape_char=escaped->length() ? (*escaped)[0] : INT_MAX;
323
300
 
324
 
  identifier::Schema identifier(*schema);
325
301
  READ_INFO read_info(file, tot_length,
326
 
                      ex->cs ? ex->cs : plugin::StorageEngine::getSchemaCollation(identifier),
327
 
                      *field_term, *ex->line_start, *ex->line_term, *enclosed,
 
302
                      ex->cs ? ex->cs : get_default_db_collation(session->db),
 
303
                      *field_term,*ex->line_start, *ex->line_term, *enclosed,
328
304
                      info.escape_char, is_fifo);
329
305
  if (read_info.error)
330
306
  {
331
307
    if  (file >= 0)
332
 
      internal::my_close(file,MYF(0));                  // no files in net reading
 
308
      my_close(file,MYF(0));                    // no files in net reading
333
309
    return(true);                               // Can't allocate buffers
334
310
  }
335
311
 
361
337
    table->next_number_field=table->found_next_number_field;
362
338
    if (ignore ||
363
339
        handle_duplicates == DUP_REPLACE)
364
 
      table->cursor->extra(HA_EXTRA_IGNORE_DUP_KEY);
 
340
      table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
365
341
    if (handle_duplicates == DUP_REPLACE)
366
 
        table->cursor->extra(HA_EXTRA_WRITE_CAN_REPLACE);
367
 
    table->cursor->ha_start_bulk_insert((ha_rows) 0);
 
342
        table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
 
343
    table->file->ha_start_bulk_insert((ha_rows) 0);
368
344
    table->copy_blobs=1;
369
345
 
370
 
    session->setAbortOnWarning(true);
 
346
    session->abort_on_warning= true;
371
347
 
372
348
    if (!field_term->length() && !enclosed->length())
373
349
      error= read_fixed_length(session, info, table_list, fields_vars,
377
353
      error= read_sep_field(session, info, table_list, fields_vars,
378
354
                            set_fields, set_values, read_info,
379
355
                            *enclosed, skip_lines, ignore);
380
 
    if (table->cursor->ha_end_bulk_insert() && !error)
 
356
    if (table->file->ha_end_bulk_insert() && !error)
381
357
    {
382
 
      table->print_error(errno, MYF(0));
 
358
      table->file->print_error(my_errno, MYF(0));
383
359
      error= 1;
384
360
    }
385
 
    table->cursor->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
386
 
    table->cursor->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
 
361
    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
362
    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
387
363
    table->next_number_field=0;
388
364
  }
389
365
  if (file >= 0)
390
 
    internal::my_close(file,MYF(0));
 
366
    my_close(file,MYF(0));
391
367
  free_blobs(table);                            /* if pack_blob was used */
392
368
  table->copy_blobs=0;
393
 
  session->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
 
369
  session->count_cuted_fields= CHECK_FIELD_IGNORE;
394
370
  /*
395
371
     simulated killing in the middle of per-row loop
396
372
     must be effective for binlogging
397
373
  */
398
 
  killed_status= (error == 0)? Session::NOT_KILLED : session->getKilled();
 
374
  killed_status= (error == 0)? Session::NOT_KILLED : session->killed;
399
375
  if (error)
400
376
  {
401
377
    error= -1;                          // Error on read
402
378
    goto err;
403
379
  }
404
 
 
405
 
  char msg[FN_REFLEN];
406
 
  snprintf(msg, sizeof(msg), ER(ER_LOAD_INFO), info.records, info.deleted,
407
 
           (info.records - info.copied), session->cuted_fields);
408
 
 
409
 
  if (session->transaction.stmt.hasModifiedNonTransData())
410
 
    session->transaction.all.markModifiedNonTransData();
 
380
  sprintf(name, ER(ER_LOAD_INFO), (uint32_t) info.records, (uint32_t) info.deleted,
 
381
          (uint32_t) (info.records - info.copied), (uint32_t) session->cuted_fields);
 
382
 
 
383
  if (session->transaction.stmt.modified_non_trans_table)
 
384
    session->transaction.all.modified_non_trans_table= true;
411
385
 
412
386
  /* ok to client sent only after binlog write and engine commit */
413
 
  session->my_ok(info.copied + info.deleted, 0, 0L, msg);
 
387
  session->my_ok(info.copied + info.deleted, 0L, name);
414
388
err:
415
389
  assert(transactional_table || !(info.copied || info.deleted) ||
416
 
              session->transaction.stmt.hasModifiedNonTransData());
417
 
  table->cursor->ha_release_auto_increment();
 
390
              session->transaction.stmt.modified_non_trans_table);
 
391
  table->file->ha_release_auto_increment();
418
392
  table->auto_increment_field_not_null= false;
419
 
  session->setAbortOnWarning(false);
420
 
 
 
393
  session->abort_on_warning= 0;
421
394
  return(error);
422
395
}
423
396
 
427
400
****************************************************************************/
428
401
 
429
402
static int
430
 
read_fixed_length(Session *session, CopyInfo &info, TableList *table_list,
 
403
read_fixed_length(Session *session, COPY_INFO &info, TableList *table_list,
431
404
                  List<Item> &fields_vars, List<Item> &set_fields,
432
405
                  List<Item> &set_values, READ_INFO &read_info,
433
406
                  uint32_t skip_lines, bool ignore_check_option_errors)
442
415
 
443
416
  while (!read_info.read_fixed_length())
444
417
  {
445
 
    if (session->getKilled())
 
418
    if (session->killed)
446
419
    {
447
420
      session->send_kill_message();
448
421
      return(1);
460
433
    }
461
434
    it.rewind();
462
435
    unsigned char *pos=read_info.row_start;
463
 
#ifdef HAVE_VALGRIND
 
436
#ifdef HAVE_purify
464
437
    read_info.row_end[0]=0;
465
438
#endif
466
439
 
487
460
        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
488
461
                            ER_WARN_TOO_FEW_RECORDS,
489
462
                            ER(ER_WARN_TOO_FEW_RECORDS), session->row_count);
490
 
 
491
 
        if (not field->maybe_null() and field->is_timestamp())
492
 
            ((field::Epoch::pointer) field)->set_time();
 
463
        if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
464
            ((Field_timestamp*) field)->set_time();
493
465
      }
494
466
      else
495
467
      {
516
488
                          ER(ER_WARN_TOO_MANY_RECORDS), session->row_count);
517
489
    }
518
490
 
519
 
    if (session->getKilled() ||
 
491
    if (session->killed ||
520
492
        fill_record(session, set_fields, set_values,
521
493
                    ignore_check_option_errors))
522
494
      return(1);
547
519
 
548
520
 
549
521
static int
550
 
read_sep_field(Session *session, CopyInfo &info, TableList *table_list,
 
522
read_sep_field(Session *session, COPY_INFO &info, TableList *table_list,
551
523
               List<Item> &fields_vars, List<Item> &set_fields,
552
524
               List<Item> &set_values, READ_INFO &read_info,
553
525
               String &enclosed, uint32_t skip_lines,
565
537
 
566
538
  for (;;it.rewind())
567
539
  {
568
 
    if (session->getKilled())
 
540
    if (session->killed)
569
541
    {
570
542
      session->send_kill_message();
571
543
      return(1);
605
577
            return(1);
606
578
          }
607
579
          field->set_null();
608
 
          if (not field->maybe_null())
 
580
          if (!field->maybe_null())
609
581
          {
610
 
            if (field->is_timestamp())
611
 
            {
612
 
              ((field::Epoch::pointer) field)->set_time();
613
 
            }
 
582
            if (field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
583
              ((Field_timestamp*) field)->set_time();
614
584
            else if (field != table->next_number_field)
615
 
            {
616
 
              field->set_warning(DRIZZLE_ERROR::WARN_LEVEL_WARN, ER_WARN_NULL_TO_NOTNULL, 1);
617
 
            }
 
585
              field->set_warning(DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
586
                                 ER_WARN_NULL_TO_NOTNULL, 1);
618
587
          }
619
588
        }
620
589
        else if (item->type() == Item::STRING_ITEM)
660
629
    }
661
630
    if (item)
662
631
    {
663
 
      /* Have not read any field, thus input cursor is simply ended */
 
632
      /* Have not read any field, thus input file is simply ended */
664
633
      if (item == fields_vars.head())
665
634
        break;
666
635
      for (; item ; item= it++)
675
644
                     session->row_count);
676
645
            return(1);
677
646
          }
678
 
          if (not field->maybe_null() and field->is_timestamp())
679
 
              ((field::Epoch::pointer) field)->set_time();
 
647
          if (!field->maybe_null() && field->type() == DRIZZLE_TYPE_TIMESTAMP)
 
648
              ((Field_timestamp*) field)->set_time();
680
649
          /*
681
650
            QQ: We probably should not throw warning for each field.
682
651
            But how about intention to always have the same number
701
670
      }
702
671
    }
703
672
 
704
 
    if (session->getKilled() ||
 
673
    if (session->killed ||
705
674
        fill_record(session, set_fields, set_values,
706
675
                    ignore_check_option_errors))
707
676
      return(1);
722
691
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_WARN,
723
692
                          ER_WARN_TOO_MANY_RECORDS, ER(ER_WARN_TOO_MANY_RECORDS),
724
693
                          session->row_count);
725
 
      if (session->getKilled())
 
694
      if (session->killed)
726
695
        return(1);
727
696
    }
728
697
    session->row_count++;
743
712
  case 'r': return '\r';
744
713
  case 'b': return '\b';
745
714
  case '0': return 0;                           // Ascii null
746
 
  case 'Z': return '\032';                      // Win32 end of cursor
 
715
  case 'Z': return '\032';                      // Win32 end of file
747
716
  case 'N': found_null=1;
748
717
 
749
718
    /* fall through */
758
727
*/
759
728
 
760
729
 
761
 
READ_INFO::READ_INFO(int file_par, size_t tot_length,
 
730
READ_INFO::READ_INFO(File file_par, size_t tot_length,
762
731
                     const CHARSET_INFO * const cs,
763
732
                     String &field_term, String &line_start, String &line_term,
764
733
                     String &enclosed_par, int escape, bool is_fifo)
765
 
  :cursor(file_par),escape_char(escape)
 
734
  :file(file_par),escape_char(escape)
766
735
{
767
736
  read_charset= cs;
768
737
  field_term_ptr=(char*) field_term.ptr();
796
765
 
797
766
 
798
767
  /* Set of a stack for unget if long terminators */
799
 
  size_t length= max(field_term_length,line_term_length)+1;
800
 
  set_if_bigger(length, line_start.length());
801
 
  stack= stack_pos= (int*) memory::sql_alloc(sizeof(int)*length);
 
768
  uint32_t length= max(field_term_length,line_term_length)+1;
 
769
  set_if_bigger(length,line_start.length());
 
770
  stack= stack_pos= (int*) sql_alloc(sizeof(int)*length);
802
771
 
803
772
  if (!(buffer=(unsigned char*) calloc(1, buff_length+1)))
804
 
    error=1;
 
773
    error=1; /* purecov: inspected */
805
774
  else
806
775
  {
807
776
    end_of_buff=buffer+buff_length;
808
 
    if (cache.init_io_cache((false) ? -1 : cursor, 0,
809
 
                            (false) ? internal::READ_NET :
810
 
                            (is_fifo ? internal::READ_FIFO : internal::READ_CACHE),0L,1,
811
 
                            MYF(MY_WME)))
 
777
    if (init_io_cache(&cache,(false) ? -1 : file, 0,
 
778
                      (false) ? READ_NET :
 
779
                      (is_fifo ? READ_FIFO : READ_CACHE),0L,1,
 
780
                      MYF(MY_WME)))
812
781
    {
813
 
      free((unsigned char*) buffer);
 
782
      free((unsigned char*) buffer); /* purecov: inspected */
814
783
      error=1;
815
784
    }
816
785
    else
831
800
  if (!error)
832
801
  {
833
802
    if (need_end_io_cache)
834
 
      cache.end_io_cache();
 
803
      ::end_io_cache(&cache);
835
804
    free(buffer);
836
805
    error=1;
837
806
  }
1030
999
 
1031
1000
  NOTES
1032
1001
    The row may not be fixed size on disk if there are escape
1033
 
    characters in the cursor.
 
1002
    characters in the file.
1034
1003
 
1035
1004
  IMPLEMENTATION NOTE
1036
1005
    One can't use fixed length with multi-byte charset **
1163
1132
}
1164
1133
 
1165
1134
 
1166
 
} /* namespace drizzled */