← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/filesystem_engine/filesystem_engine.cc

Added the testsuite location finding code to support in-plugin-dir test suites.

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/filesystem_engine/filesystem_engine.cc
1
 
/*
2
 
  Copyright (C) 2010 Zimin
3
 
 
4
 
  This program is free software; you can redistribute it and/or
5
 
  modify it under the terms of the GNU General Public License
6
 
  as published by the Free Software Foundation; either version 2
7
 
  of the License, or (at your option) any later version.
8
 
 
9
 
  This program is distributed in the hope that it will be useful,
10
 
  but WITHOUT ANY WARRANTY; without even the implied warranty of
11
 
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12
 
  GNU General Public License for more details.
13
 
 
14
 
  You should have received a copy of the GNU General Public License
15
 
  along with this program; if not, write to the Free Software
16
 
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
17
 
*/
18
 
 
19
 
#include "config.h"
20
 
#include <drizzled/field.h>
21
 
#include <drizzled/field/blob.h>
22
 
#include <drizzled/field/timestamp.h>
23
 
#include <drizzled/error.h>
24
 
#include <drizzled/table.h>
25
 
#include <drizzled/session.h>
26
 
#include "drizzled/internal/my_sys.h"
27
 
#include <google/protobuf/io/zero_copy_stream.h>
28
 
#include <google/protobuf/io/zero_copy_stream_impl.h>
29
 
 
30
 
#include "filesystem_engine.h"
31
 
#include "utility.h"
32
 
 
33
 
#include <fcntl.h>
34
 
 
35
 
#include <string>
36
 
#include <map>
37
 
#include <fstream>
38
 
#include <sstream>
39
 
#include <iostream>
40
 
#include <boost/algorithm/string.hpp>
41
 
 
42
 
using namespace std;
43
 
using namespace drizzled;
44
 
 
45
 
#define FILESYSTEM_EXT ".FST"
46
 
 
47
 
/* Stuff for shares */
48
 
pthread_mutex_t filesystem_mutex;
49
 
 
50
 
static const char *ha_filesystem_exts[] = {
51
 
  FILESYSTEM_EXT,
52
 
  NULL
53
 
};
54
 
 
55
 
class FilesystemEngine : public drizzled::plugin::StorageEngine
56
 
{
57
 
private:
58
 
  typedef std::map<string, FilesystemTableShare*> FilesystemMap;
59
 
  FilesystemMap fs_open_tables;
60
 
public:
61
 
  FilesystemEngine(const string& name_arg)
62
 
   : drizzled::plugin::StorageEngine(name_arg,
63
 
                                     HTON_NULL_IN_KEY |
64
 
                                     HTON_SKIP_STORE_LOCK |
65
 
                                     HTON_CAN_INDEX_BLOBS |
66
 
                                     HTON_AUTO_PART_KEY),
67
 
     fs_open_tables()
68
 
  {
69
 
    table_definition_ext= FILESYSTEM_EXT;
70
 
    pthread_mutex_init(&filesystem_mutex, MY_MUTEX_INIT_FAST);
71
 
  }
72
 
  virtual ~FilesystemEngine()
73
 
  {
74
 
    pthread_mutex_destroy(&filesystem_mutex);
75
 
  }
76
 
 
77
 
  virtual Cursor *create(Table &table)
78
 
  {
79
 
    return new FilesystemCursor(*this, table);
80
 
  }
81
 
 
82
 
  const char **bas_ext() const {
83
 
    return ha_filesystem_exts;
84
 
  }
85
 
 
86
 
  bool validateCreateTableOption(const std::string &key, const std::string &state);
87
 
 
88
 
  int doCreateTable(Session &,
89
 
                    Table &table_arg,
90
 
                    const drizzled::TableIdentifier &identifier,
91
 
                    drizzled::message::Table&);
92
 
 
93
 
  int doGetTableDefinition(Session& ,
94
 
                           const drizzled::TableIdentifier &,
95
 
                           drizzled::message::Table &);
96
 
 
97
 
  int doDropTable(Session&, const TableIdentifier &);
98
 
 
99
 
  /* operations on FilesystemTableShare */
100
 
  FilesystemTableShare *findOpenTable(const string table_name);
101
 
  void addOpenTable(const string &table_name, FilesystemTableShare *);
102
 
  void deleteOpenTable(const string &table_name);
103
 
 
104
 
  uint32_t max_keys()          const { return 0; }
105
 
  uint32_t max_key_parts()     const { return 0; }
106
 
  uint32_t max_key_length()    const { return 0; }
107
 
  bool doDoesTableExist(Session& , const TableIdentifier &);
108
 
  int doRenameTable(Session&, const TableIdentifier &, const TableIdentifier &);
109
 
  void doGetTableIdentifiers(drizzled::CachedDirectory &directory,
110
 
                             const drizzled::SchemaIdentifier &schema_identifier,
111
 
                             drizzled::TableIdentifier::vector &set_of_identifiers);
112
 
private:
113
 
  void getTableNamesFromFilesystem(drizzled::CachedDirectory &directory,
114
 
                                   const drizzled::SchemaIdentifier &schema_identifier,
115
 
                                   drizzled::plugin::TableNameList *set_of_names,
116
 
                                   drizzled::TableIdentifier::vector *set_of_identifiers);
117
 
};
118
 
 
119
 
void FilesystemEngine::getTableNamesFromFilesystem(drizzled::CachedDirectory &directory,
120
 
                                                   const drizzled::SchemaIdentifier &schema_identifier,
121
 
                                                   drizzled::plugin::TableNameList *set_of_names,
122
 
                                                   drizzled::TableIdentifier::vector *set_of_identifiers)
123
 
{
124
 
  drizzled::CachedDirectory::Entries entries= directory.getEntries();
125
 
 
126
 
  for (drizzled::CachedDirectory::Entries::iterator entry_iter= entries.begin();
127
 
      entry_iter != entries.end();
128
 
      ++entry_iter)
129
 
  {
130
 
    drizzled::CachedDirectory::Entry *entry= *entry_iter;
131
 
    const string *filename= &entry->filename;
132
 
 
133
 
    assert(not filename->empty());
134
 
 
135
 
    string::size_type suffix_pos= filename->rfind('.');
136
 
 
137
 
    if (suffix_pos != string::npos &&
138
 
        boost::iequals(filename->substr(suffix_pos), FILESYSTEM_EXT) &&
139
 
        filename->compare(0, strlen(TMP_FILE_PREFIX), TMP_FILE_PREFIX))
140
 
    {
141
 
      char uname[NAME_LEN + 1];
142
 
      uint32_t file_name_len;
143
 
 
144
 
      file_name_len= TableIdentifier::filename_to_tablename(filename->c_str(), uname, sizeof(uname));
145
 
      uname[file_name_len - sizeof(FILESYSTEM_EXT) + 1]= '\0';
146
 
      if (set_of_names)
147
 
        set_of_names->insert(uname);
148
 
      if (set_of_identifiers)
149
 
        set_of_identifiers->push_back(TableIdentifier(schema_identifier, uname));
150
 
    }
151
 
  }
152
 
}
153
 
 
154
 
void FilesystemEngine::doGetTableIdentifiers(drizzled::CachedDirectory &directory,
155
 
                                             const drizzled::SchemaIdentifier &schema_identifier,
156
 
                                             drizzled::TableIdentifier::vector &set_of_identifiers)
157
 
{
158
 
  getTableNamesFromFilesystem(directory, schema_identifier, NULL, &set_of_identifiers);
159
 
}
160
 
 
161
 
int FilesystemEngine::doDropTable(Session &, const TableIdentifier &identifier)
162
 
{
163
 
  string new_path(identifier.getPath());
164
 
  new_path+= FILESYSTEM_EXT;
165
 
  int err= unlink(new_path.c_str());
166
 
  if (err)
167
 
  {
168
 
    err= errno;
169
 
  }
170
 
  return err;
171
 
}
172
 
 
173
 
bool FilesystemEngine::doDoesTableExist(Session &, const TableIdentifier &identifier)
174
 
{
175
 
  string proto_path(identifier.getPath());
176
 
  proto_path.append(FILESYSTEM_EXT);
177
 
 
178
 
  if (access(proto_path.c_str(), F_OK))
179
 
  {
180
 
    return false;
181
 
  }
182
 
 
183
 
  return true;
184
 
}
185
 
 
186
 
FilesystemTableShare *FilesystemEngine::findOpenTable(const string table_name)
187
 
{
188
 
  FilesystemMap::iterator find_iter=
189
 
    fs_open_tables.find(table_name);
190
 
 
191
 
  if (find_iter != fs_open_tables.end())
192
 
    return (*find_iter).second;
193
 
  else
194
 
    return NULL;
195
 
}
196
 
 
197
 
void FilesystemEngine::addOpenTable(const string &table_name, FilesystemTableShare *share)
198
 
{
199
 
  fs_open_tables[table_name]= share;
200
 
}
201
 
 
202
 
void FilesystemEngine::deleteOpenTable(const string &table_name)
203
 
{
204
 
  fs_open_tables.erase(table_name);
205
 
}
206
 
 
207
 
static int parseTaggedFile(const FormatInfo &fi, vector< map<string, string> > &v)
208
 
{
209
 
  int filedesc= ::open(fi.getFileName().c_str(), O_RDONLY);
210
 
  if (filedesc < 0)
211
 
    return errno;
212
 
 
213
 
  boost::scoped_ptr<TransparentFile> filebuffer(new TransparentFile);
214
 
  filebuffer->init_buff(filedesc);
215
 
 
216
 
  bool last_line_empty= false;
217
 
  map<string, string> kv;
218
 
  int pos= 0;
219
 
  string line;
220
 
  while (1)
221
 
  {
222
 
    char ch= filebuffer->get_value(pos);
223
 
    if (ch == '\0')
224
 
    {
225
 
      if (!last_line_empty)
226
 
      {
227
 
        v.push_back(kv);
228
 
        kv.clear();
229
 
      }
230
 
      break;
231
 
    }
232
 
    ++pos;
233
 
 
234
 
    if (!fi.isRowSeparator(ch))
235
 
    {
236
 
      line.push_back(ch);
237
 
      continue;
238
 
    }
239
 
 
240
 
    // if we have a new empty line,
241
 
    // it means we got the end of a section, push it to vector
242
 
    if (line.empty())
243
 
    {
244
 
      if (!last_line_empty)
245
 
      {
246
 
        v.push_back(kv);
247
 
        kv.clear();
248
 
      }
249
 
      last_line_empty= true;
250
 
      continue;
251
 
    }
252
 
 
253
 
    // parse the line
254
 
    vector<string> sv, svcopy;
255
 
    boost::split(sv, line, boost::is_any_of(fi.getColSeparator()));
256
 
    for (vector<string>::iterator iter= sv.begin();
257
 
         iter != sv.end();
258
 
         ++iter)
259
 
    {
260
 
      if (!iter->empty())
261
 
        svcopy.push_back(*iter);
262
 
    }
263
 
 
264
 
    // the first splitted string as key,
265
 
    // and the second splitted string as value.
266
 
    string key(svcopy[0]);
267
 
    boost::trim(key);
268
 
    if (svcopy.size() >= 2)
269
 
    {
270
 
      string value(svcopy[1]);
271
 
      boost::trim(value);
272
 
      kv[key]= value;
273
 
    }
274
 
    else if (svcopy.size() >= 1)
275
 
      kv[key]= "";
276
 
 
277
 
    last_line_empty= false;
278
 
    line.clear();
279
 
  }
280
 
  close(filedesc);
281
 
  return 0;
282
 
}
283
 
 
284
 
int FilesystemEngine::doGetTableDefinition(Session &,
285
 
                                           const drizzled::TableIdentifier &identifier,
286
 
                                           drizzled::message::Table &table_proto)
287
 
{
288
 
  string new_path(identifier.getPath());
289
 
  new_path.append(FILESYSTEM_EXT);
290
 
 
291
 
  int fd= ::open(new_path.c_str(), O_RDONLY);
292
 
  if (fd < 0)
293
 
    return ENOENT;
294
 
 
295
 
  google::protobuf::io::ZeroCopyInputStream* input=
296
 
    new google::protobuf::io::FileInputStream(fd);
297
 
 
298
 
  if (not input)
299
 
    return HA_ERR_CRASHED_ON_USAGE;
300
 
 
301
 
  if (not table_proto.ParseFromZeroCopyStream(input))
302
 
  {
303
 
    close(fd);
304
 
    delete input;
305
 
    if (not table_proto.IsInitialized())
306
 
    {
307
 
      my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),
308
 
               table_proto.InitializationErrorString().c_str());
309
 
      return ER_CORRUPT_TABLE_DEFINITION;
310
 
    }
311
 
 
312
 
    return HA_ERR_CRASHED_ON_USAGE;
313
 
  }
314
 
  delete input;
315
 
 
316
 
  // if the file is a tagged file such as /proc/meminfo
317
 
  // then columns of this table are added dynamically here.
318
 
  FormatInfo format;
319
 
  format.parseFromTable(&table_proto);
320
 
  if (not format.isTagFormat() || not format.isFileGiven())
321
 
  {
322
 
    close(fd);
323
 
    return EEXIST;
324
 
  }
325
 
 
326
 
  std::vector< std::map<std::string, std::string> > vm;
327
 
  if (parseTaggedFile(format, vm) != 0)
328
 
  {
329
 
    close(fd);
330
 
 
331
 
    return EEXIST;
332
 
  }
333
 
  if (vm.size() == 0) {
334
 
    close(fd);
335
 
    return EEXIST;
336
 
  }
337
 
 
338
 
  // we don't care what user provides, just clear them all
339
 
  table_proto.clear_field();
340
 
  // we take the first section as sample
341
 
  std::map<string, string> kv= vm[0];
342
 
  for (std::map<string, string>::iterator iter= kv.begin();
343
 
       iter != kv.end();
344
 
       ++iter)
345
 
  {
346
 
    // add columns to table proto
347
 
    message::Table::Field *field= table_proto.add_field();
348
 
    field->set_name(iter->first);
349
 
    field->set_type(drizzled::message::Table::Field::VARCHAR);
350
 
    message::Table::Field::StringFieldOptions *stringoption= field->mutable_string_options();
351
 
    stringoption->set_length(iter->second.length() + 1);
352
 
  }
353
 
 
354
 
  close(fd);
355
 
  return EEXIST;
356
 
}
357
 
 
358
 
FilesystemTableShare::FilesystemTableShare(const string table_name_arg)
359
 
  : use_count(0), table_name(table_name_arg),
360
 
  update_file_opened(false),
361
 
  needs_reopen(false)
362
 
{
363
 
}
364
 
 
365
 
FilesystemTableShare::~FilesystemTableShare()
366
 
{
367
 
  pthread_mutex_destroy(&mutex);
368
 
}
369
 
 
370
 
FilesystemTableShare *FilesystemCursor::get_share(const char *table_name)
371
 
{
372
 
  Guard g(filesystem_mutex);
373
 
 
374
 
  FilesystemEngine *a_engine= static_cast<FilesystemEngine *>(getEngine());
375
 
  share= a_engine->findOpenTable(table_name);
376
 
 
377
 
  /*
378
 
    If share is not present in the hash, create a new share and
379
 
    initialize its members.
380
 
  */
381
 
  if (share == NULL)
382
 
  {
383
 
    share= new (nothrow) FilesystemTableShare(table_name);
384
 
    if (share == NULL)
385
 
    {
386
 
      return NULL;
387
 
    }
388
 
 
389
 
    share->format.parseFromTable(getTable()->getShare()->getTableProto());
390
 
    if (!share->format.isFileGiven())
391
 
    {
392
 
      return NULL;
393
 
    }
394
 
    /*
395
 
     * for taggered file such as /proc/meminfo,
396
 
     * we pre-process it first, and store the parsing result in a map.
397
 
     */
398
 
    if (share->format.isTagFormat())
399
 
    {
400
 
      if (parseTaggedFile(share->format, share->vm) != 0)
401
 
      {
402
 
        return NULL;
403
 
      }
404
 
    }
405
 
    a_engine->addOpenTable(share->table_name, share);
406
 
 
407
 
    pthread_mutex_init(&share->mutex, MY_MUTEX_INIT_FAST);
408
 
  }
409
 
  share->use_count++;
410
 
 
411
 
  return share;
412
 
}
413
 
 
414
 
void FilesystemCursor::free_share()
415
 
{
416
 
  Guard g(filesystem_mutex);
417
 
 
418
 
  if (!--share->use_count){
419
 
    FilesystemEngine *a_engine= static_cast<FilesystemEngine *>(getEngine());
420
 
    a_engine->deleteOpenTable(share->table_name);
421
 
    pthread_mutex_destroy(&share->mutex);
422
 
    delete share;
423
 
  }
424
 
}
425
 
 
426
 
void FilesystemCursor::critical_section_enter()
427
 
{
428
 
  if (sql_command_type == SQLCOM_ALTER_TABLE ||
429
 
      sql_command_type == SQLCOM_UPDATE ||
430
 
      sql_command_type == SQLCOM_DELETE ||
431
 
      sql_command_type == SQLCOM_INSERT ||
432
 
      sql_command_type == SQLCOM_INSERT_SELECT ||
433
 
      sql_command_type == SQLCOM_REPLACE ||
434
 
      sql_command_type == SQLCOM_REPLACE_SELECT)
435
 
    share->filesystem_lock.scan_update_begin();
436
 
  else
437
 
    share->filesystem_lock.scan_begin();
438
 
 
439
 
  thread_locked = true;
440
 
}
441
 
 
442
 
void FilesystemCursor::critical_section_exit()
443
 
{
444
 
  if (sql_command_type == SQLCOM_ALTER_TABLE ||
445
 
      sql_command_type == SQLCOM_UPDATE ||
446
 
      sql_command_type == SQLCOM_DELETE ||
447
 
      sql_command_type == SQLCOM_INSERT ||
448
 
      sql_command_type == SQLCOM_INSERT_SELECT ||
449
 
      sql_command_type == SQLCOM_REPLACE ||
450
 
      sql_command_type == SQLCOM_REPLACE_SELECT)
451
 
    share->filesystem_lock.scan_update_end();
452
 
  else
453
 
    share->filesystem_lock.scan_end();
454
 
 
455
 
  thread_locked = false;
456
 
}
457
 
 
458
 
FilesystemCursor::FilesystemCursor(drizzled::plugin::StorageEngine &engine_arg, Table &table_arg)
459
 
  : Cursor(engine_arg, table_arg),
460
 
    file_buff(new TransparentFile),
461
 
    thread_locked(false)
462
 
{
463
 
}
464
 
 
465
 
int FilesystemCursor::doOpen(const drizzled::TableIdentifier &identifier, int, uint32_t)
466
 
{
467
 
  if (!(share= get_share(identifier.getPath().c_str())))
468
 
    return ENOENT;
469
 
 
470
 
  file_desc= ::open(share->format.getFileName().c_str(), O_RDONLY);
471
 
  if (file_desc < 0)
472
 
  {
473
 
    free_share();
474
 
    return ER_CANT_OPEN_FILE;
475
 
  }
476
 
 
477
 
  ref_length= sizeof(off_t);
478
 
  return 0;
479
 
}
480
 
 
481
 
int FilesystemCursor::close(void)
482
 
{
483
 
  int err= ::close(file_desc);
484
 
  if (err < 0)
485
 
    err= errno;
486
 
  free_share();
487
 
  return err;
488
 
}
489
 
 
490
 
int FilesystemCursor::doStartTableScan(bool)
491
 
{
492
 
  sql_command_type = session_sql_command(getTable()->getSession());
493
 
 
494
 
  if (thread_locked)
495
 
    critical_section_exit();
496
 
  critical_section_enter();
497
 
 
498
 
  if (share->format.isTagFormat())
499
 
  {
500
 
    tag_depth= 0;
501
 
    return 0;
502
 
  }
503
 
 
504
 
  current_position= 0;
505
 
  next_position= 0;
506
 
  slots.clear();
507
 
  if (share->needs_reopen)
508
 
  {
509
 
    file_desc= ::open(share->format.getFileName().c_str(), O_RDONLY);
510
 
    if (file_desc < 0)
511
 
      return HA_ERR_CRASHED_ON_USAGE;
512
 
    share->needs_reopen= false;
513
 
  }
514
 
  file_buff->init_buff(file_desc);
515
 
  return 0;
516
 
}
517
 
 
518
 
int FilesystemCursor::find_current_row(unsigned char *buf)
519
 
{
520
 
  ptrdiff_t row_offset= buf - getTable()->record[0];
521
 
 
522
 
  next_position= current_position;
523
 
 
524
 
  string content;
525
 
  bool line_done= false;
526
 
  bool line_blank= true;
527
 
  Field **field= getTable()->getFields();
528
 
  for (; !line_done && *field; ++next_position)
529
 
  {
530
 
    char ch= file_buff->get_value(next_position);
531
 
    if (ch == '\0')
532
 
      return HA_ERR_END_OF_FILE;
533
 
 
534
 
    if (share->format.isEscapedChar(ch))
535
 
    {
536
 
      // read next character
537
 
      ch= file_buff->get_value(++next_position);
538
 
      if (ch == '\0')
539
 
        return HA_ERR_END_OF_FILE;
540
 
 
541
 
      content.push_back(FormatInfo::getEscapedChar(ch));
542
 
 
543
 
      continue;
544
 
    }
545
 
 
546
 
    // if we find separator
547
 
    bool is_row= share->format.isRowSeparator(ch);
548
 
    bool is_col= share->format.isColSeparator(ch);
549
 
    if (content.empty())
550
 
    {
551
 
      if (share->format.isSeparatorModeGeneral() && is_row && line_blank)
552
 
        continue;
553
 
      if (share->format.isSeparatorModeWeak() && is_col)
554
 
        continue;
555
 
    }
556
 
 
557
 
    if (is_row || is_col)
558
 
    {
559
 
      (*field)->move_field_offset(row_offset);
560
 
      if (!content.empty())
561
 
      {
562
 
        (*field)->set_notnull();
563
 
        if ((*field)->isReadSet() || (*field)->isWriteSet())
564
 
        {
565
 
          (*field)->setWriteSet();
566
 
          (*field)->store_and_check(CHECK_FIELD_WARN,
567
 
                                    content.c_str(),
568
 
                                    (uint32_t)content.length(),
569
 
                                    &my_charset_bin);
570
 
        }
571
 
        else
572
 
        {
573
 
          (*field)->set_default();
574
 
        }
575
 
      }
576
 
      else
577
 
        (*field)->set_null();
578
 
      (*field)->move_field_offset(-row_offset);
579
 
 
580
 
      content.clear();
581
 
      ++field;
582
 
 
583
 
      line_blank= false;
584
 
      if (is_row)
585
 
        line_done= true;
586
 
 
587
 
      continue;
588
 
    }
589
 
    content.push_back(ch);
590
 
  }
591
 
  if (line_done)
592
 
  {
593
 
    for (; *field; ++field)
594
 
    {
595
 
      (*field)->move_field_offset(row_offset);
596
 
      (*field)->set_notnull();
597
 
      (*field)->set_default();
598
 
      (*field)->move_field_offset(-row_offset);
599
 
    }
600
 
  }
601
 
  else
602
 
  {
603
 
    // eat up characters when line_done
604
 
    while (!line_done)
605
 
    {
606
 
      char ch= file_buff->get_value(next_position);
607
 
      if (share->format.isRowSeparator(ch))
608
 
        line_done= true;
609
 
      ++next_position;
610
 
    }
611
 
  }
612
 
  return 0;
613
 
}
614
 
 
615
 
int FilesystemCursor::rnd_next(unsigned char *buf)
616
 
{
617
 
  ha_statistic_increment(&system_status_var::ha_read_rnd_next_count);
618
 
  if (share->format.isTagFormat())
619
 
  {
620
 
    if (tag_depth >= share->vm.size())
621
 
      return HA_ERR_END_OF_FILE;
622
 
 
623
 
    ptrdiff_t row_offset= buf - getTable()->record[0];
624
 
    for (Field **field= getTable()->getFields(); *field; field++)
625
 
    {
626
 
      string key((*field)->field_name);
627
 
      string content= share->vm[tag_depth][key];
628
 
 
629
 
      (*field)->move_field_offset(row_offset);
630
 
      if (!content.empty())
631
 
      {
632
 
        (*field)->set_notnull();
633
 
        if ((*field)->isReadSet() || (*field)->isWriteSet())
634
 
        {
635
 
          (*field)->setWriteSet();
636
 
          (*field)->store_and_check(CHECK_FIELD_WARN,
637
 
                                    content.c_str(),
638
 
                                    (uint32_t)content.length(),
639
 
                                    &my_charset_bin);
640
 
        }
641
 
        else
642
 
        {
643
 
          (*field)->set_default();
644
 
        }
645
 
      }
646
 
      else
647
 
      {
648
 
        (*field)->set_null();
649
 
      }
650
 
      (*field)->move_field_offset(-row_offset);
651
 
    }
652
 
    ++tag_depth;
653
 
    return 0;
654
 
  }
655
 
  // normal file
656
 
  current_position= next_position;
657
 
  return find_current_row(buf);
658
 
}
659
 
 
660
 
void FilesystemCursor::position(const unsigned char *)
661
 
{
662
 
  *reinterpret_cast<off_t *>(ref)= current_position;
663
 
}
664
 
 
665
 
int FilesystemCursor::rnd_pos(unsigned char * buf, unsigned char *pos)
666
 
{
667
 
  ha_statistic_increment(&system_status_var::ha_read_rnd_count);
668
 
  current_position= *reinterpret_cast<off_t *>(pos);
669
 
  return find_current_row(buf);
670
 
}
671
 
 
672
 
int FilesystemCursor::info(uint32_t)
673
 
{
674
 
  if (stats.records < 2)
675
 
    stats.records= 2;
676
 
  return 0;
677
 
}
678
 
 
679
 
int FilesystemCursor::openUpdateFile()
680
 
{
681
 
  if (!share->update_file_opened)
682
 
  {
683
 
    struct stat st;
684
 
    if (stat(share->format.getFileName().c_str(), &st) < 0)
685
 
      return -1;
686
 
    update_file_name= share->format.getFileName();
687
 
    update_file_name.append(".UPDATE");
688
 
    unlink(update_file_name.c_str());
689
 
    update_file_desc= ::open(update_file_name.c_str(),
690
 
                             O_RDWR | O_CREAT | O_TRUNC,
691
 
                             st.st_mode);
692
 
    if (update_file_desc < 0)
693
 
    {
694
 
      return -1;
695
 
    }
696
 
    share->update_file_opened= true;
697
 
  }
698
 
  return 0;
699
 
}
700
 
 
701
 
int FilesystemCursor::doEndTableScan()
702
 
{
703
 
  sql_command_type = session_sql_command(getTable()->getSession());
704
 
 
705
 
  if (share->format.isTagFormat())
706
 
  {
707
 
    if (thread_locked)
708
 
      critical_section_exit();
709
 
    return 0;
710
 
  }
711
 
 
712
 
  if (slots.size() == 0)
713
 
  {
714
 
    if (thread_locked)
715
 
      critical_section_exit();
716
 
    return 0;
717
 
  }
718
 
 
719
 
  int err= -1;
720
 
  sort(slots.begin(), slots.end());
721
 
  vector< pair<off_t, off_t> >::iterator slot_iter= slots.begin();
722
 
  off_t write_start= 0;
723
 
  off_t write_end= 0;
724
 
  off_t file_buffer_start= 0;
725
 
 
726
 
  pthread_mutex_lock(&share->mutex);
727
 
 
728
 
  file_buff->init_buff(file_desc);
729
 
  if (openUpdateFile() < 0)
730
 
    goto error;
731
 
 
732
 
  while (file_buffer_start != -1)
733
 
  {
734
 
    bool in_hole= false;
735
 
 
736
 
    write_end= file_buff->end();
737
 
    if (slot_iter != slots.end() &&
738
 
      write_end >= slot_iter->first)
739
 
    {
740
 
      write_end= slot_iter->first;
741
 
      in_hole= true;
742
 
    }
743
 
 
744
 
    off_t write_length= write_end - write_start;
745
 
    if (write_in_all(update_file_desc,
746
 
               file_buff->ptr() + (write_start - file_buff->start()),
747
 
               write_length) != write_length)
748
 
      goto error;
749
 
 
750
 
    if (in_hole)
751
 
    {
752
 
      while (file_buff->end() <= slot_iter->second && file_buffer_start != -1)
753
 
        file_buffer_start= file_buff->read_next();
754
 
      write_start= slot_iter->second;
755
 
      ++slot_iter;
756
 
    }
757
 
    else
758
 
      write_start= write_end;
759
 
 
760
 
    if (write_end == file_buff->end())
761
 
      file_buffer_start= file_buff->read_next();
762
 
  }
763
 
  // close update file
764
 
  if (::fsync(update_file_desc) || 
765
 
      ::close(update_file_desc))
766
 
    goto error;
767
 
  share->update_file_opened= false;
768
 
 
769
 
  // close current file
770
 
  if (::close(file_desc))
771
 
    goto error;
772
 
  if (::rename(update_file_name.c_str(), share->format.getFileName().c_str()))
773
 
    goto error;
774
 
 
775
 
  share->needs_reopen= true;
776
 
 
777
 
error:
778
 
  err= errno;
779
 
  pthread_mutex_unlock(&share->mutex);
780
 
 
781
 
  if (thread_locked)
782
 
    critical_section_exit();
783
 
 
784
 
  return err;
785
 
}
786
 
 
787
 
void FilesystemCursor::recordToString(string& output)
788
 
{
789
 
  bool first= true;
790
 
  drizzled::String attribute;
791
 
  for (Field **field= getTable()->getFields(); *field; ++field)
792
 
  {
793
 
    if (first == true)
794
 
    {
795
 
      first= false;
796
 
    }
797
 
    else
798
 
    {
799
 
      output.append(share->format.getColSeparatorHead());
800
 
    }
801
 
 
802
 
    if (not (*field)->is_null())
803
 
    {
804
 
      (*field)->setReadSet();
805
 
      (*field)->val_str(&attribute, &attribute);
806
 
 
807
 
      output.append(attribute.ptr(), attribute.length());
808
 
    }
809
 
    else
810
 
    {
811
 
      output.append("0");
812
 
    }
813
 
  }
814
 
  output.append(share->format.getRowSeparatorHead());
815
 
}
816
 
 
817
 
int FilesystemCursor::doInsertRecord(unsigned char * buf)
818
 
{
819
 
  (void)buf;
820
 
 
821
 
  if (share->format.isTagFormat())
822
 
    return 0;
823
 
 
824
 
  sql_command_type = session_sql_command(getTable()->getSession());
825
 
 
826
 
  critical_section_enter();
827
 
 
828
 
  int err_write= 0;
829
 
  int err_close= 0;
830
 
 
831
 
  string output_line;
832
 
  recordToString(output_line);
833
 
 
834
 
  int fd= ::open(share->format.getFileName().c_str(), O_WRONLY | O_APPEND);
835
 
  if (fd < 0)
836
 
  {
837
 
    critical_section_exit();
838
 
    return ENOENT;
839
 
  }
840
 
 
841
 
  err_write= write_in_all(fd, output_line.c_str(), output_line.length());
842
 
  if (err_write < 0)
843
 
    err_write= errno;
844
 
  else
845
 
    err_write= 0;
846
 
 
847
 
  err_close= ::close(fd);
848
 
  if (err_close < 0)
849
 
    err_close= errno;
850
 
 
851
 
  critical_section_exit();
852
 
 
853
 
  if (err_write)
854
 
    return err_write;
855
 
  if (err_close)
856
 
    return err_close;
857
 
  return 0;
858
 
}
859
 
 
860
 
int FilesystemCursor::doUpdateRecord(const unsigned char *, unsigned char *)
861
 
{
862
 
  if (share->format.isTagFormat())
863
 
    return 0;
864
 
  if (openUpdateFile())
865
 
    return errno;
866
 
 
867
 
  // get the update information
868
 
  string str;
869
 
  recordToString(str);
870
 
 
871
 
  if (write_in_all(update_file_desc, str.c_str(), str.length()) < 0)
872
 
    return errno;
873
 
 
874
 
  addSlot();
875
 
 
876
 
  return 0;
877
 
}
878
 
 
879
 
void FilesystemCursor::addSlot()
880
 
{
881
 
  if (slots.size() > 0 && slots.back().second == current_position)
882
 
    slots.back().second= next_position;
883
 
  else
884
 
    slots.push_back(make_pair(current_position, next_position));
885
 
}
886
 
 
887
 
int FilesystemCursor::doDeleteRecord(const unsigned char *)
888
 
{
889
 
  if (share->format.isTagFormat())
890
 
    return 0;
891
 
  addSlot();
892
 
  return 0;
893
 
}
894
 
 
895
 
int FilesystemEngine::doRenameTable(Session&, const TableIdentifier &from, const TableIdentifier &to)
896
 
{
897
 
  if (rename_file_ext(from.getPath().c_str(), to.getPath().c_str(), FILESYSTEM_EXT))
898
 
    return errno;
899
 
  return 0;
900
 
}
901
 
 
902
 
bool FilesystemEngine::validateCreateTableOption(const std::string &key,
903
 
                                                 const std::string &state)
904
 
{
905
 
  return FormatInfo::validateOption(key, state);
906
 
}
907
 
 
908
 
int FilesystemEngine::doCreateTable(Session &,
909
 
                        Table&,
910
 
                        const drizzled::TableIdentifier &identifier,
911
 
                        drizzled::message::Table &proto)
912
 
{
913
 
  FormatInfo format;
914
 
  format.parseFromTable(&proto);
915
 
  if (format.isFileGiven())
916
 
  {
917
 
    int err= ::open(format.getFileName().c_str(), O_RDONLY);
918
 
    if (err < 0)
919
 
      return errno;
920
 
  }
921
 
 
922
 
  string new_path(identifier.getPath());
923
 
  new_path+= FILESYSTEM_EXT;
924
 
  fstream output(new_path.c_str(), ios::out | ios::binary);
925
 
 
926
 
  if (! output)
927
 
    return 1;
928
 
 
929
 
  if (! proto.SerializeToOstream(&output))
930
 
  {
931
 
    output.close();
932
 
    unlink(new_path.c_str());
933
 
    return 1;
934
 
  }
935
 
 
936
 
  return 0;
937
 
}
938
 
 
939
 
static FilesystemEngine *filesystem_engine= NULL;
940
 
 
941
 
static int filesystem_init_func(drizzled::module::Context &context)
942
 
{
943
 
  filesystem_engine= new FilesystemEngine("FILESYSTEM");
944
 
  context.add(filesystem_engine);
945
 
 
946
 
  return 0;
947
 
}
948
 
 
949
 
DRIZZLE_DECLARE_PLUGIN
950
 
{
951
 
  DRIZZLE_VERSION_ID,
952
 
  "FILESYSTEM",
953
 
  "1.0",
954
 
  "Zimin",
955
 
  "Filesystem Engine",
956
 
  PLUGIN_LICENSE_GPL,
957
 
  filesystem_init_func, /* Plugin Init */
958
 
  NULL,                       /* system variables                */
959
 
  NULL                        /* config options                  */
960
 
}
961
 
DRIZZLE_DECLARE_PLUGIN_END;