← Back to branch summary

~barry-leslie/drizzle/drizzle_pbmsV2

« back to all changes in this revision

Viewing changes to plugin/filesystem_engine/filesystem_engine.cc

  • Committer: Barry.Leslie at PrimeBase
  • Date: 2011-04-04 14:40:45 UTC
  • mfrom: (1919.2.344 build)
  • Revision ID: barry.leslie@primebase.com-20110404144045-etbtnq340cgy2dw3
Merged in lp:drizzle.

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