← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/plugin/storage_engine.cc

  • Committer: Stewart Smith
  • Date: 2010-03-30 13:10:50 UTC
  • mfrom: (1283.34.4)
  • mto: (1283.37.1)
  • mto: This revision was merged to the branch mainline in revision 1475.
  • Revision ID: stewart@flamingspork.com-20100330131050-b3hwar8bihn1s1qf
Merged embedded-innodb-full-index-read into embedded-innodb-position-rnd-pos.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/plugin/storage_engine.cc
25
25
#include <string>
26
26
#include <vector>
27
27
#include <set>
 
28
#include <fstream>
28
29
#include <algorithm>
29
30
#include <functional>
30
31
 
74
75
static std::set<std::string> set_of_table_definition_ext;
75
76
 
76
77
StorageEngine::StorageEngine(const string name_arg,
77
 
                             const bitset<HTON_BIT_SIZE> &flags_arg)
78
 
    : Plugin(name_arg, "StorageEngine"),
79
 
      MonitoredInTransaction(), /* This gives the storage engine a "slot" or ID */
80
 
      flags(flags_arg)
 
78
                             const bitset<HTON_BIT_SIZE> &flags_arg) :
 
79
  Plugin(name_arg, "StorageEngine"),
 
80
  MonitoredInTransaction(), /* This gives the storage engine a "slot" or ID */
 
81
  flags(flags_arg)
81
82
{
82
 
  pthread_mutex_init(&proto_cache_mutex, NULL);
83
83
}
84
84
 
85
85
StorageEngine::~StorageEngine()
86
86
{
87
 
  pthread_mutex_destroy(&proto_cache_mutex);
88
87
}
89
88
 
90
89
void StorageEngine::setTransactionReadWrite(Session& session)
93
92
  statement_ctx.markModifiedNonTransData();
94
93
}
95
94
 
96
 
int StorageEngine::doRenameTable(Session *,
97
 
                                 const char *from,
98
 
                                 const char *to)
 
95
 
 
96
int StorageEngine::renameTable(Session &session, TableIdentifier &from, TableIdentifier &to)
99
97
{
100
 
  int error= 0;
101
 
  for (const char **ext= bas_ext(); *ext ; ext++)
102
 
  {
103
 
    if (rename_file_ext(from, to, *ext))
104
 
    {
105
 
      if ((error=errno) != ENOENT)
106
 
        break;
107
 
      error= 0;
108
 
    }
109
 
  }
110
 
  return error;
 
98
  setTransactionReadWrite(session);
 
99
 
 
100
  return doRenameTable(session, from, to);
111
101
}
112
102
 
113
 
 
114
103
/**
115
104
  Delete all files with extension from bas_ext().
116
105
 
143
132
        break;
144
133
    }
145
134
    else
 
135
    {
146
136
      enoent_or_zero= 0;                        // No error for ENOENT
 
137
    }
 
138
 
147
139
    error= enoent_or_zero;
148
140
  }
149
141
  return error;
184
176
class FindEngineByName
185
177
  : public unary_function<StorageEngine *, bool>
186
178
{
187
 
  const string target;
 
179
  const string &target;
 
180
 
188
181
public:
189
 
  explicit FindEngineByName(const string target_arg)
190
 
    : target(target_arg)
191
 
  {}
 
182
  explicit FindEngineByName(const string &target_arg) :
 
183
    target(target_arg)
 
184
  {
 
185
  }
192
186
  result_type operator() (argument_type engine)
193
187
  {
194
188
    string engine_name(engine->getName());
199
193
  }
200
194
};
201
195
 
202
 
StorageEngine *StorageEngine::findByName(string find_str)
 
196
StorageEngine *StorageEngine::findByName(const string &find_str)
203
197
{
204
 
  transform(find_str.begin(), find_str.end(),
205
 
            find_str.begin(), ::tolower);
 
198
  string search_string(find_str);
 
199
  transform(search_string.begin(), search_string.end(),
 
200
            search_string.begin(), ::tolower);
206
201
 
207
202
  
208
203
  EngineVector::iterator iter= find_if(vector_of_engines.begin(),
209
204
                                       vector_of_engines.end(),
210
 
                                       FindEngineByName(find_str));
 
205
                                       FindEngineByName(search_string));
211
206
  if (iter != vector_of_engines.end())
212
207
  {
213
208
    StorageEngine *engine= *iter;
218
213
  return NULL;
219
214
}
220
215
 
221
 
StorageEngine *StorageEngine::findByName(Session& session, string find_str)
 
216
StorageEngine *StorageEngine::findByName(Session& session, const string &find_str)
222
217
{
223
 
  
224
 
  transform(find_str.begin(), find_str.end(),
225
 
            find_str.begin(), ::tolower);
 
218
  string search_string(find_str);
 
219
  transform(search_string.begin(), search_string.end(),
 
220
            search_string.begin(), ::tolower);
226
221
 
227
 
  if (find_str.compare("default") == 0)
 
222
  if (search_string.compare("default") == 0)
228
223
    return session.getDefaultStorageEngine();
229
224
 
230
225
  EngineVector::iterator iter= find_if(vector_of_engines.begin(),
231
226
                                       vector_of_engines.end(),
232
 
                                       FindEngineByName(find_str));
 
227
                                       FindEngineByName(search_string));
233
228
  if (iter != vector_of_engines.end())
234
229
  {
235
230
    StorageEngine *engine= *iter;
240
235
  return NULL;
241
236
}
242
237
 
243
 
class StorageEngineCloseConnection
244
 
: public unary_function<StorageEngine *, void>
 
238
class StorageEngineCloseConnection : public unary_function<StorageEngine *, void>
245
239
{
246
240
  Session *session;
247
241
public:
303
297
 
304
298
  result_type operator() (argument_type engine)
305
299
  {
306
 
    int ret= engine->doGetTableDefinition(session,
307
 
                                          identifier,
308
 
                                          table_message);
 
300
    int ret= engine->doGetTableDefinition(session, identifier, table_message);
309
301
 
310
302
    if (ret != ENOENT)
311
303
      err= ret;
458
450
  int error= 0;
459
451
  int error_proto;
460
452
  message::Table src_proto;
461
 
  StorageEngine* engine;
 
453
  StorageEngine *engine;
462
454
 
463
455
  error_proto= StorageEngine::getTableDefinition(session, identifier, src_proto);
464
456
 
465
457
  if (error_proto == ER_CORRUPT_TABLE_DEFINITION)
466
458
  {
467
 
    my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),
468
 
             src_proto.InitializationErrorString().c_str());
 
459
    string error_message;
 
460
 
 
461
    error_message.append(identifier.getSQLPath());
 
462
    error_message.append(" : ");
 
463
    error_message.append(src_proto.InitializationErrorString());
 
464
 
 
465
    my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0), error_message.c_str());
 
466
 
469
467
    return ER_CORRUPT_TABLE_DEFINITION;
470
468
  }
471
469
 
473
471
 
474
472
  if (engine)
475
473
  {
476
 
    std::string path(identifier.getPath());
477
 
    engine->setTransactionReadWrite(session);
478
 
    error= engine->doDropTable(session, identifier);
479
 
 
480
 
    if (not error)
481
 
    {
482
 
      if (not engine->check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
483
 
      {
484
 
        uint64_t counter; // @todo We need to refactor to check that.
485
 
 
486
 
        for_each(vector_of_schema_engines.begin(), vector_of_schema_engines.end(),
487
 
                 DropTable(session, identifier, counter));
488
 
      }
489
 
    }
 
474
    error= StorageEngine::dropTable(session, *engine, identifier);
490
475
  }
491
476
 
492
477
  if (error_proto && error == 0)
495
480
  return error;
496
481
}
497
482
 
 
483
int StorageEngine::dropTable(Session& session,
 
484
                             StorageEngine &engine,
 
485
                             TableIdentifier &identifier)
 
486
{
 
487
  int error;
 
488
 
 
489
  engine.setTransactionReadWrite(session);
 
490
  error= engine.doDropTable(session, identifier);
 
491
 
 
492
  if (not error)
 
493
  {
 
494
    if (not engine.check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
 
495
    {
 
496
      uint64_t counter; // @todo We need to refactor to check that.
 
497
 
 
498
      for_each(vector_of_schema_engines.begin(), vector_of_schema_engines.end(),
 
499
               DropTable(session, identifier, counter));
 
500
    }
 
501
  }
 
502
 
 
503
  return error;
 
504
}
 
505
 
498
506
/**
499
507
  Initiates table-file and calls appropriate database-creator.
500
508
 
502
510
   0  ok
503
511
  @retval
504
512
   1  error
505
 
 
506
 
   @todo refactor to remove goto
507
513
*/
508
 
int StorageEngine::createTable(Session& session,
 
514
int StorageEngine::createTable(Session &session,
509
515
                               TableIdentifier &identifier,
510
516
                               bool update_create_info,
511
517
                               message::Table& table_message)
515
521
  TableShare share(identifier.getDBName().c_str(), 0, identifier.getTableName().c_str(), identifier.getPath().c_str());
516
522
  message::Table tmp_proto;
517
523
 
518
 
  if (parse_table_proto(session, table_message, &share))
519
 
    goto err;
520
 
 
521
 
  if (open_table_from_share(&session, &share, "", 0, 0,
522
 
                            &table))
523
 
    goto err;
524
 
 
525
 
  if (update_create_info)
526
 
    table.updateCreateInfo(&table_message);
527
 
 
528
 
  /* Check for legal operations against the Engine using the proto (if used) */
529
 
  if (table_message.type() == message::Table::TEMPORARY &&
530
 
      share.storage_engine->check_flag(HTON_BIT_TEMPORARY_NOT_SUPPORTED) == true)
531
 
  {
532
 
    error= HA_ERR_UNSUPPORTED;
533
 
    goto err2;
534
 
  }
535
 
  else if (table_message.type() != message::Table::TEMPORARY &&
536
 
           share.storage_engine->check_flag(HTON_BIT_TEMPORARY_ONLY) == true)
537
 
  {
538
 
    error= HA_ERR_UNSUPPORTED;
539
 
    goto err2;
540
 
  }
541
 
 
542
 
  {
543
 
    if (not share.storage_engine->check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
544
 
    {
545
 
      int protoerr= StorageEngine::writeDefinitionFromPath(identifier, table_message);
546
 
 
547
 
      if (protoerr)
548
 
      {
549
 
        error= protoerr;
550
 
        goto err2;
551
 
      }
552
 
    }
553
 
 
554
 
    share.storage_engine->setTransactionReadWrite(session);
555
 
 
556
 
    error= share.storage_engine->doCreateTable(&session,
557
 
                                               table,
558
 
                                               identifier,
559
 
                                               table_message);
560
 
  }
561
 
 
562
 
err2:
563
 
  if (error)
564
 
  {
565
 
    if (not share.storage_engine->check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
566
 
      plugin::StorageEngine::deleteDefinitionFromPath(identifier);
567
 
 
568
 
    my_error(ER_CANT_CREATE_TABLE, MYF(ME_BELL+ME_WAITTANG), identifier.getSQLPath().c_str(), error);
569
 
  }
570
 
 
571
 
  table.closefrm(false);
572
 
 
573
 
err:
 
524
  if (parse_table_proto(session, table_message, &share) || open_table_from_share(&session, &share, "", 0, 0, &table))
 
525
  { 
 
526
    // @note Error occured, we should probably do a little more here.
 
527
  }
 
528
  else
 
529
  {
 
530
    if (update_create_info)
 
531
      table.updateCreateInfo(&table_message);
 
532
 
 
533
    /* Check for legal operations against the Engine using the proto (if used) */
 
534
    if (table_message.type() == message::Table::TEMPORARY &&
 
535
        share.storage_engine->check_flag(HTON_BIT_TEMPORARY_NOT_SUPPORTED) == true)
 
536
    {
 
537
      error= HA_ERR_UNSUPPORTED;
 
538
    }
 
539
    else if (table_message.type() != message::Table::TEMPORARY &&
 
540
             share.storage_engine->check_flag(HTON_BIT_TEMPORARY_ONLY) == true)
 
541
    {
 
542
      error= HA_ERR_UNSUPPORTED;
 
543
    }
 
544
    else
 
545
    {
 
546
      bool do_create= true;
 
547
      if (not share.storage_engine->check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
 
548
      {
 
549
        int protoerr= StorageEngine::writeDefinitionFromPath(identifier, table_message);
 
550
 
 
551
        if (protoerr)
 
552
        {
 
553
          error= protoerr;
 
554
          do_create= false;
 
555
        }
 
556
      }
 
557
 
 
558
      if (do_create)
 
559
      {
 
560
        share.storage_engine->setTransactionReadWrite(session);
 
561
 
 
562
        error= share.storage_engine->doCreateTable(&session,
 
563
                                                   table,
 
564
                                                   identifier,
 
565
                                                   table_message);
 
566
      }
 
567
    }
 
568
 
 
569
    if (error)
 
570
    {
 
571
      if (not share.storage_engine->check_flag(HTON_BIT_HAS_DATA_DICTIONARY))
 
572
        plugin::StorageEngine::deleteDefinitionFromPath(identifier);
 
573
 
 
574
      my_error(ER_CANT_CREATE_TABLE, MYF(ME_BELL+ME_WAITTANG), identifier.getSQLPath().c_str(), error);
 
575
    }
 
576
 
 
577
    table.closefrm(false);
 
578
  }
 
579
 
574
580
  share.free_table_share();
575
581
  return(error != 0);
576
582
}
638
644
 
639
645
class StorageEngineGetSchemaDefinition: public unary_function<StorageEngine *, bool>
640
646
{
641
 
  const std::string &schema_name;
 
647
  std::string schema_name;
642
648
  message::Schema &schema_proto;
643
649
 
644
650
public:
645
 
  StorageEngineGetSchemaDefinition(const std::string &schema_name_arg,
 
651
  StorageEngineGetSchemaDefinition(const std::string schema_name_arg,
646
652
                                  message::Schema &schema_proto_arg) :
647
653
    schema_name(schema_name_arg),
648
654
    schema_proto(schema_proto_arg) 
649
 
  { }
 
655
  {
 
656
    transform(schema_name.begin(), schema_name.end(),
 
657
              schema_name.begin(), ::tolower);
 
658
  }
650
659
 
651
660
  result_type operator() (argument_type engine)
652
661
  {
657
666
/*
658
667
  Return value is "if parsed"
659
668
*/
 
669
bool StorageEngine::getSchemaDefinition(TableIdentifier &identifier, message::Schema &proto)
 
670
{
 
671
  return StorageEngine::getSchemaDefinition(identifier.getSchemaName(), proto);
 
672
}
 
673
 
660
674
bool StorageEngine::getSchemaDefinition(const std::string &schema_name, message::Schema &proto)
661
675
{
662
676
  proto.Clear();
680
694
  return StorageEngine::getSchemaDefinition(schema_name, proto);
681
695
}
682
696
 
 
697
bool StorageEngine::doesSchemaExist(TableIdentifier &identifier)
 
698
{
 
699
  message::Schema proto;
 
700
 
 
701
  return StorageEngine::getSchemaDefinition(identifier.getSchemaName(), proto);
 
702
}
 
703
 
683
704
 
684
705
const CHARSET_INFO *StorageEngine::getSchemaCollation(const std::string &schema_name)
685
706
{
843
864
class DropTables: public unary_function<StorageEngine *, void>
844
865
{
845
866
  Session &session;
846
 
  TableNameList &set_of_names;
 
867
  TableIdentifierList &table_identifiers;
847
868
 
848
869
public:
849
870
 
850
 
  DropTables(Session &session_arg, set<string>& of_names) :
 
871
  DropTables(Session &session_arg, TableIdentifierList &table_identifiers_arg) :
851
872
    session(session_arg),
852
 
    set_of_names(of_names)
 
873
    table_identifiers(table_identifiers_arg)
853
874
  { }
854
875
 
855
876
  result_type operator() (argument_type engine)
856
877
  {
857
 
    for (TableNameList::iterator iter= set_of_names.begin();
858
 
         iter != set_of_names.end();
 
878
    for (TableIdentifierList::iterator iter= table_identifiers.begin();
 
879
         iter != table_identifiers.end();
859
880
         iter++)
860
881
    {
861
 
      TableIdentifier dummy((*iter).c_str());
862
 
      int error= engine->doDropTable(session, dummy);
 
882
      int error= engine->doDropTable(session, const_cast<TableIdentifier&>(*iter));
863
883
 
864
884
      // On a return of zero we know we found and deleted the table. So we
865
885
      // remove it from our search.
866
886
      if (not error)
867
 
        set_of_names.erase(iter);
 
887
        table_identifiers.erase(iter);
868
888
    }
869
889
  }
870
890
};
877
897
void StorageEngine::removeLostTemporaryTables(Session &session, const char *directory)
878
898
{
879
899
  CachedDirectory dir(directory, set_of_table_definition_ext);
880
 
  set<string> set_of_table_names;
 
900
  TableIdentifierList table_identifiers;
881
901
 
882
902
  if (dir.fail())
883
903
  {
903
923
    path+= directory;
904
924
    path+= FN_LIBCHAR;
905
925
    path+= entry->filename;
906
 
    set_of_table_names.insert(path);
 
926
    message::Table definition;
 
927
    if (StorageEngine::readTableFile(path, definition))
 
928
    {
 
929
      TableIdentifier identifier(definition.schema(), definition.name(), path);
 
930
      table_identifiers.push_back(identifier);
 
931
    }
907
932
  }
908
933
 
909
934
  for_each(vector_of_engines.begin(), vector_of_engines.end(),
910
 
           DropTables(session, set_of_table_names));
 
935
           DropTables(session, table_identifiers));
911
936
  
912
937
  /*
913
938
    Now we just clean up anything that might left over.
1213
1238
 
1214
1239
int StorageEngine::renameDefinitionFromPath(TableIdentifier &dest, TableIdentifier &src)
1215
1240
{
 
1241
  message::Table table_message;
1216
1242
  string src_path(src.getPath());
1217
1243
  string dest_path(dest.getPath());
1218
1244
 
1219
1245
  src_path.append(DEFAULT_DEFINITION_FILE_EXT);
1220
1246
  dest_path.append(DEFAULT_DEFINITION_FILE_EXT);
1221
1247
 
1222
 
  return internal::my_rename(src_path.c_str(), dest_path.c_str(), MYF(MY_WME));
 
1248
  bool was_read= StorageEngine::readTableFile(src_path.c_str(), table_message);
 
1249
 
 
1250
  if (not was_read)
 
1251
  {
 
1252
    return ENOENT;
 
1253
  }
 
1254
 
 
1255
  dest.copyToTableMessage(table_message);
 
1256
 
 
1257
  int error= StorageEngine::writeDefinitionFromPath(dest, table_message);
 
1258
 
 
1259
  if (not error)
 
1260
  {
 
1261
    if (unlink(src_path.c_str()))
 
1262
      perror(src_path.c_str());
 
1263
  }
 
1264
 
 
1265
  return error;
1223
1266
}
1224
1267
 
1225
1268
int StorageEngine::writeDefinitionFromPath(TableIdentifier &identifier, message::Table &table_message)
1226
1269
{
 
1270
  char definition_file_tmp[FN_REFLEN];
1227
1271
  string file_name(identifier.getPath());
1228
1272
 
1229
1273
  file_name.append(DEFAULT_DEFINITION_FILE_EXT);
1230
1274
 
1231
 
  int fd= open(file_name.c_str(), O_RDWR|O_CREAT|O_TRUNC, internal::my_umask);
 
1275
  snprintf(definition_file_tmp, sizeof(definition_file_tmp), "%s.%sXXXXXX", file_name.c_str(), DEFAULT_DEFINITION_FILE_EXT.c_str());
 
1276
 
 
1277
  int fd= mkstemp(definition_file_tmp);
1232
1278
 
1233
1279
  if (fd == -1)
 
1280
  {
 
1281
    perror(definition_file_tmp);
1234
1282
    return errno;
 
1283
  }
1235
1284
 
1236
1285
  google::protobuf::io::ZeroCopyOutputStream* output=
1237
1286
    new google::protobuf::io::FileOutputStream(fd);
1238
1287
 
1239
 
  if (table_message.SerializeToZeroCopyStream(output) == false)
 
1288
  if (not table_message.SerializeToZeroCopyStream(output))
1240
1289
  {
 
1290
    my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),
 
1291
             table_message.InitializationErrorString().c_str());
1241
1292
    delete output;
1242
 
    close(fd);
1243
 
    return errno;
 
1293
 
 
1294
    if (close(fd) == -1)
 
1295
      perror(definition_file_tmp);
 
1296
 
 
1297
    if (unlink(definition_file_tmp) == -1)
 
1298
      perror(definition_file_tmp);
 
1299
 
 
1300
    return ER_CORRUPT_TABLE_DEFINITION;
1244
1301
  }
1245
1302
 
1246
1303
  delete output;
1247
 
  close(fd);
 
1304
 
 
1305
  if (close(fd) == -1)
 
1306
  {
 
1307
    int error= errno;
 
1308
    perror(definition_file_tmp);
 
1309
 
 
1310
    if (unlink(definition_file_tmp))
 
1311
      perror(definition_file_tmp);
 
1312
 
 
1313
    return error;
 
1314
  }
 
1315
 
 
1316
  if (rename(definition_file_tmp, file_name.c_str()) == -1)
 
1317
  {
 
1318
    int error= errno;
 
1319
    perror(definition_file_tmp);
 
1320
 
 
1321
    if (unlink(definition_file_tmp))
 
1322
      perror(definition_file_tmp);
 
1323
 
 
1324
    return error;
 
1325
  }
 
1326
 
1248
1327
  return 0;
1249
1328
}
1250
1329
 
1281
1360
  return false;
1282
1361
}
1283
1362
 
 
1363
bool StorageEngine::readTableFile(const std::string &path, message::Table &table_message)
 
1364
{
 
1365
  fstream input(path.c_str(), ios::in | ios::binary);
 
1366
 
 
1367
  if (input.good())
 
1368
  {
 
1369
    if (table_message.ParseFromIstream(&input))
 
1370
    {
 
1371
      return true;
 
1372
    }
 
1373
 
 
1374
    my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),
 
1375
             table_message.InitializationErrorString().c_str());
 
1376
  }
 
1377
  else
 
1378
  {
 
1379
    perror(path.c_str());
 
1380
  }
 
1381
 
 
1382
  return false;
 
1383
}
 
1384
 
 
1385
 
 
1386
 
1284
1387
} /* namespace plugin */
1285
1388
} /* namespace drizzled */