1
/* Copyright (C) 2008 PrimeBase Technologies GmbH, Germany
3
* PrimeBase Media Stream for MySQL
5
* This program is free software; you can redistribute it and/or modify
6
* it under the terms of the GNU General Public License as published by
7
* the Free Software Foundation; either version 2 of the License, or
8
* (at your option) any later version.
10
* This program is distributed in the hope that it will be useful,
11
* but WITHOUT ANY WARRANTY; without even the implied warranty of
12
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
* GNU General Public License for more details.
15
* You should have received a copy of the GNU General Public License
16
* along with this program; if not, write to the Free Software
17
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
* Original author: Paul McCullagh
20
* Continued development: Barry Leslie
32
#include <drizzled/common.h>
33
#include <drizzled/current_session.h>
34
#include <drizzled/session.h>
38
#include "cslib/CSConfig.h"
39
#include "cslib/CSGlobal.h"
40
#include "cslib/CSStrUtil.h"
41
#include "cslib/CSThread.h"
44
#define PBMS_API pbms_internal
48
#include "engine_ms.h"
49
#include "connection_handler_ms.h"
50
#include "open_table_ms.h"
51
#include "network_ms.h"
52
#include "transaction_ms.h"
61
extern CSThread *pbms_getMySelf(THD *thd);
62
extern void pbms_setMySelf(THD *thd, CSThread *self);
67
* ---------------------------------------------------------------
68
* ENGINE CALL-IN INTERFACE
71
static PBMS_API *StreamingEngines;
72
// If PBMS support is built directly into the mysql/drizzle handler code
73
// then calls from all other handlers are ignored.
74
static bool have_handler_support = false;
77
* ---------------------------------------------------------------
78
* ENGINE CALLBACK INTERFACE
81
static void ms_register_engine(PBMSEnginePtr engine)
83
if (engine->ms_internal)
84
have_handler_support = true;
87
static void ms_deregister_engine(PBMSEnginePtr engine)
92
static int ms_create_blob(bool internal, const char *db_name, const char *tab_name, char *blob, size_t blob_len, PBMSBlobURLPtr blob_url, PBMSResultPtr result)
94
if (have_handler_support && !internal) {
95
MSEngine::errorResult(CS_CONTEXT, MS_ERR_INVALID_OPERATION, "Invalid ms_create_blob() call", result);
99
return MSEngine::createBlob(db_name, tab_name, blob, blob_len, blob_url, result);
103
* ms_use_blob() may or may not alter the blob url depending on the type of URL and if the BLOB is in a
104
* different database or not. It may also add a BLOB reference to the BLOB table log if the BLOB was from
105
* a different table or no table was specified when the BLOB was uploaded.
107
* There is no need to undo this function because it will be undone automaticly if the BLOB is not retained.
109
static int ms_retain_blob(bool internal, const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, unsigned short col_index, PBMSResultPtr result)
111
if (have_handler_support && !internal) {
112
cs_strcpy(PBMS_BLOB_URL_SIZE, ret_blob_url->bu_data, blob_url); // This should have already been converted.
116
return MSEngine::referenceBlob(db_name, tab_name, ret_blob_url, blob_url, col_index, result);
119
static int ms_release_blob(bool internal, const char *db_name, const char *tab_name, char *blob_url, PBMSResultPtr result)
122
if (have_handler_support && !internal)
125
return MSEngine::dereferenceBlob(db_name, tab_name, blob_url, result);
128
static int ms_drop_table(bool internal, const char *db_name, const char *tab_name, PBMSResultPtr result)
130
if (have_handler_support && !internal)
133
return MSEngine::dropTable(db_name, tab_name, result);
136
static int ms_rename_table(bool internal, const char * db_name, const char *from_table, const char *to_db, const char *to_table, PBMSResultPtr result)
138
if (have_handler_support && !internal)
141
return MSEngine::renameTable(db_name, from_table, to_db, to_table, result);
144
static void ms_completed(bool internal, bool ok)
146
if (have_handler_support && !internal)
149
MSEngine::callCompleted(ok);
152
PBMSCallbacksRec engine_callbacks = {
155
ms_deregister_engine,
164
// =============================
165
int MSEngine::startUp(PBMSResultPtr result)
169
StreamingEngines = new PBMS_API();
170
err = StreamingEngines->PBMSStartup(&engine_callbacks, result);
172
delete StreamingEngines;
173
else { // Register the PBMS enabled engines the startup before PBMS
174
PBMSSharedMemoryPtr sh_mem = StreamingEngines->sharedMemory;
175
PBMSEnginePtr engine;
177
for (int i=0; i<sh_mem->sm_list_len; i++) {
178
if ((engine = sh_mem->sm_engine_list[i]))
179
ms_register_engine(engine);
185
void MSEngine::shutDown()
187
StreamingEngines->PBMSShutdown();
189
delete StreamingEngines;
192
const PBMSEnginePtr MSEngine::getEngineInfoAt(int indx)
194
PBMSSharedMemoryPtr sh_mem = StreamingEngines->sharedMemory;
195
PBMSEnginePtr engine = NULL;
198
for (int i=0; i<sh_mem->sm_list_len; i++) {
199
if ((engine = sh_mem->sm_engine_list[i])) {
207
return (const PBMSEnginePtr)NULL;
212
bool MSEngine::try_createBlob(CSThread *self, const char *db_name, const char *tab_name, char *blob, size_t blob_len, PBMSBlobURLPtr blob_url)
214
volatile bool rtc = true;
218
CSInputStream *i_stream = NULL;
220
otab = openTable(db_name, tab_name, true);
223
if (!otab->getDB()->isRecovering()) {
224
i_stream = CSMemoryInputStream::newStream((unsigned char *)blob, blob_len);
225
otab->createBlob(blob_url, blob_len, NULL, 0, i_stream);
227
CSException::throwException(CS_CONTEXT, MS_ERR_RECOVERY_IN_PROGRESS, "Cannot create BLOBs during repository recovery.");
238
int32_t MSEngine::createBlob(const char *db_name, const char *tab_name, char *blob, size_t blob_len, PBMSBlobURLPtr blob_url, PBMSResultPtr result)
244
if ((err = enterConnectionNoThd(&self, result)))
248
if (try_createBlob(self, db_name, tab_name, blob, blob_len, blob_url))
249
err = exceptionToResult(&self->myException, result);
255
bool MSEngine::try_referenceBlob(CSThread *self, const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, uint16_t col_index)
257
volatile bool rtc = true;
262
if (! PBMSBlobURLTools::couldBeURL(blob_url, &blob)){
263
char buffer[CS_EXC_MESSAGE_SIZE];
265
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Incorrect URL: ");
266
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, blob_url);
267
CSException::throwException(CS_CONTEXT, MS_ERR_INCORRECT_URL, buffer);
270
otab = openTable(db_name, tab_name, true);
273
otab->useBlob(blob.bu_type, blob.bu_db_id, blob.bu_tab_id, blob.bu_blob_id, blob.bu_auth_code, col_index, blob.bu_blob_size, blob.bu_blob_ref_id, ret_blob_url);
284
int32_t MSEngine::referenceBlob(const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, uint16_t col_index, PBMSResultPtr result)
290
if ((err = enterConnectionNoThd(&self, result)))
294
if (try_referenceBlob(self, db_name, tab_name, ret_blob_url, blob_url, col_index))
295
err = exceptionToResult(&self->myException, result);
302
bool MSEngine::try_dereferenceBlob(CSThread *self, const char *db_name, const char *tab_name, char *blob_url)
304
volatile bool rtc = true;
309
if (! PBMSBlobURLTools::couldBeURL(blob_url, &blob)){
310
char buffer[CS_EXC_MESSAGE_SIZE];
312
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Incorrect URL: ");
313
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, blob_url);
314
CSException::throwException(CS_CONTEXT, MS_ERR_INCORRECT_URL, buffer);
317
otab = openTable(db_name, tab_name, true);
319
if (!otab->getDB()->isRecovering()) {
320
if (otab->getTableID() == blob.bu_tab_id)
321
otab->releaseReference(blob.bu_blob_id, blob.bu_blob_ref_id);
323
char buffer[CS_EXC_MESSAGE_SIZE];
325
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Incorrect table ID: ");
326
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, blob_url);
327
CSException::throwException(CS_CONTEXT, MS_ERR_INCORRECT_URL, buffer);
331
char buffer[CS_EXC_MESSAGE_SIZE];
333
cs_strcpy(CS_EXC_MESSAGE_SIZE, buffer, "Incorrect URL: ");
334
cs_strcat(CS_EXC_MESSAGE_SIZE, buffer, blob_url);
335
CSException::throwException(CS_CONTEXT, MS_ERR_INCORRECT_URL, buffer);
346
int32_t MSEngine::dereferenceBlob(const char *db_name, const char *tab_name, char *blob_url, PBMSResultPtr result)
351
if ((err = enterConnectionNoThd(&self, result)))
355
if (try_dereferenceBlob(self, db_name, tab_name, blob_url))
356
err = exceptionToResult(&self->myException, result);
361
bool MSEngine::try_dropDatabase(CSThread *self, const char *db_name)
363
volatile bool rtc = true;
365
MSDatabase::dropDatabase(db_name);
374
int32_t MSEngine::dropDatabase(const char *db_name, PBMSResultPtr result)
379
if ((err = enterConnectionNoThd(&self, result)))
384
if (try_dropDatabase(self, db_name))
385
err = exceptionToResult(&self->myException, result);
391
typedef struct UnDoInfo {
393
CSString *udo_toDatabaseName;
394
CSString *udo_fromDatabaseName;
395
CSString *udo_OldName;
396
CSString *udo_NewName;
397
} UnDoInfoRec, *UnDoInfoPtr;
400
bool MSEngine::try_dropTable(CSThread *self, const char *db_name, const char *tab_name)
402
volatile bool rtc = true;
408
MSOpenTablePool *tab_pool;
410
UnDoInfoPtr undo_info = NULL;
412
undo_info = (UnDoInfoPtr) cs_malloc(sizeof(UnDoInfoRec));
414
undo_info->udo_WasRename = false;
415
self->myInfo = undo_info;
417
otab = openTable(db_name, tab_name, false);
422
// If we are recovering do not delete the table.
423
// It is normal for MySQL recovery scripts to delete any table they aare about to
424
// recover and then recreate it. If this is done after the repository has been recovered
425
// then this would delete all the recovered BLOBs in the table.
426
if (otab->getDB()->isRecovering()) {
427
otab->returnToPool();
433
// Before dropping the table the table ref file is renamed so that
434
// it is out of the way incase a new table is created before the
435
// old one is cleaned up.
437
old_path = otab->getDBTable()->getTableFile();
440
new_path = otab->getDBTable()->getTableFile(tab_name, true);
442
// Rearrage the object stack to pop the otab object
450
tab = otab->getDBTable();
454
tab_pool = MSTableList::lockTablePoolForDeletion(otab);
457
if (old_path->exists())
458
old_path->move(RETAIN(new_path));
459
tab->myDatabase->dropTable(RETAIN(tab));
461
/* Add the table to the temp delete list if we are not recovering... */
462
tab->prepareToDelete();
464
backtopool_(tab_pool); // The will unlock and close the table pool freeing all tables in it.
465
pop_(tab); // Returning the pool will have released this. (YUK!)
480
int32_t MSEngine::dropTable(const char *db_name, const char *tab_name, PBMSResultPtr result)
485
if ((err = enterConnectionNoThd(&self, result)))
489
if (try_dropTable(self, db_name, tab_name))
490
err = exceptionToResult(&self->myException, result);
498
static void completeDeleteTable(UnDoInfoPtr info, bool ok)
500
// TO DO: figure out a way to undo the delete.
503
CSException::throwException(CS_CONTEXT, MS_ERR_NOT_IMPLEMENTED, "Cannot undo delete table.");
507
bool MSEngine::renameTable(const char *from_db_name, const char *from_table, const char *to_db_name, const char *to_table)
512
MSOpenTablePool *tab_pool;
517
if (strcmp(to_db_name, from_db_name) != 0) {
518
CSException::throwException(CS_CONTEXT, MS_ERR_NOT_IMPLEMENTED, "Cannot rename tables containing BLOBs across databases (yet). Sorry!");
521
otab = openTable(from_db_name, from_table, false);
527
if (otab->getDB()->isRecovering())
528
CSException::throwException(CS_CONTEXT, MS_ERR_RECOVERY_IN_PROGRESS, "Cannot rename tables during repository recovery.");
530
from_path = otab->getDBTable()->getTableFile();
533
to_path = otab->getDBTable()->getTableFile(to_table, false);
535
// Rearrage the object stack to pop the otab object
543
otab->openForReading();
544
tab = otab->getDBTable();
549
tab_pool = MSTableList::lockTablePoolForDeletion(otab);
552
from_path->move(RETAIN(to_path));
553
tab->myDatabase->renameTable(tab, to_table);
555
backtopool_(tab_pool); // The will unlock and close the table pool freeing all tables in it.
556
pop_(tab); // Returning the pool will have released this. (YUK!)
564
bool MSEngine::try_renameTable(CSThread *self, const char *from_db_name, const char *from_table, const char *to_db_name, const char *to_table)
566
volatile bool rtc = true;
568
UnDoInfoPtr undo_info = (UnDoInfoPtr) cs_malloc(sizeof(UnDoInfoRec));
569
push_ptr_(undo_info);
571
undo_info->udo_WasRename = true;
572
if (renameTable(from_db_name, from_table, to_db_name, to_table)) {
573
undo_info->udo_fromDatabaseName = CSString::newString(from_db_name);
574
push_(undo_info->udo_fromDatabaseName);
576
undo_info->udo_toDatabaseName = CSString::newString(to_db_name);
577
push_(undo_info->udo_toDatabaseName);
579
undo_info->udo_OldName = CSString::newString(from_table);
580
push_(undo_info->udo_OldName);
582
undo_info->udo_NewName = CSString::newString(to_table);
584
pop_(undo_info->udo_OldName);
585
pop_(undo_info->udo_toDatabaseName);
586
pop_(undo_info->udo_fromDatabaseName);
588
undo_info->udo_fromDatabaseName = undo_info->udo_toDatabaseName = undo_info->udo_OldName = undo_info->udo_NewName = NULL;
590
self->myInfo = undo_info;
600
int32_t MSEngine::renameTable(const char *from_db_name, const char *from_table, const char *to_db_name, const char *to_table, PBMSResultPtr result)
605
if ((err = enterConnectionNoThd(&self, result)))
609
if (try_renameTable(self, from_db_name, from_table, to_db_name, to_table))
610
err = exceptionToResult(&self->myException, result);
618
void MSEngine::completeRenameTable(UnDoInfoPtr info, bool ok)
620
// Swap the paths around here to revers the rename.
621
CSString *from_db_name= info->udo_toDatabaseName;
622
CSString *to_db_name= info->udo_fromDatabaseName;
623
CSString *from_table= info->udo_NewName;
624
CSString *to_table= info->udo_OldName;
635
renameTable(from_db_name->getCString(), from_table->getCString(), to_db_name->getCString(), to_table->getCString());
638
release_(to_db_name);
639
release_(from_table);
640
release_(from_db_name);
646
static bool try_CompleteTransaction(CSThread *self, bool ok)
648
volatile bool rtc = true;
651
MSTransactionManager::commit();
652
else if (self->myIsAutoCommit)
653
MSTransactionManager::rollback();
655
MSTransactionManager::rollbackToPosition(self->myStartStmt); // Rollback the last logical statement.
665
void MSEngine::callCompleted(bool ok)
668
PBMSResultRec result;
670
if (enterConnectionNoThd(&self, &result))
674
UnDoInfoPtr info = (UnDoInfoPtr) self->myInfo;
675
if (info->udo_WasRename)
676
completeRenameTable(info, ok);
678
completeDeleteTable(info, ok);
682
} else if (self->myTID && (self->myIsAutoCommit || !ok)) {
684
if (try_CompleteTransaction(self, ok)) {
685
self->logException();
690
self->myStartStmt = self->myStmtCount;
694
MSOpenTable *MSEngine::openTable(const char *db_name, const char *tab_name, bool create)
696
MSOpenTable *otab = NULL;
697
uint32_t db_id, tab_id;
700
if ( MSDatabase::convertTableAndDatabaseToIDs(db_name, tab_name, &db_id, &tab_id, create))
701
otab = MSTableList::getOpenTableByID(db_id, tab_id);
707
bool MSEngine::couldBeURL(const char *blob_url, size_t length)
710
return PBMSBlobURLTools::couldBeURL(blob_url, length, &blob);
714
int MSEngine::exceptionToResult(CSException *e, PBMSResultPtr result)
716
const char *context, *trace;
718
result->mr_code = e->getErrorCode();
719
cs_strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, e->getMessage());
720
context = e->getContext();
721
trace = e->getStackTrace();
722
if (context && *context) {
723
cs_strcpy(MS_RESULT_STACK_SIZE, result->mr_stack, context);
725
cs_strcat(MS_RESULT_STACK_SIZE, result->mr_stack, "\n");
728
*result->mr_stack = 0;
730
cs_strcat(MS_RESULT_STACK_SIZE, result->mr_stack, trace);
731
return MS_ERR_ENGINE;
735
int MSEngine::errorResult(const char *func, const char *file, int line, int err, const char *message, PBMSResultPtr result)
739
e.initException(func, file, line, err, message);
740
return exceptionToResult(&e, result);
744
int MSEngine::osErrorResult(const char *func, const char *file, int line, int err, PBMSResultPtr result)
748
e.initOSError(func, file, line, err);
749
return MSEngine::exceptionToResult(&e, result);
753
int MSEngine::enterConnection(THD *thd, CSThread **r_self, PBMSResultPtr result, bool doCreate)
755
CSThread *self = NULL;
758
// In drizzle there is no 1:1 relationship between pthreads and sessions
759
// so we must always get it from the session handle NOT the current pthread.
760
self = CSThread::getSelf();
764
if (!(self = pbms_getMySelf(thd))) {
766
return MS_ERR_NOT_FOUND;
768
if (!(self = CSThread::newCSThread()))
769
return osErrorResult(CS_CONTEXT, ENOMEM, result);
770
if (!CSThread::attach(self))
771
return MSEngine::exceptionToResult(&self->myException, result);
772
pbms_setMySelf(thd, self);
774
if (!CSThread::setSelf(self))
775
return MSEngine::exceptionToResult(&self->myException, result);
779
return MS_ERR_NOT_FOUND;
781
if (!(self = CSThread::newCSThread()))
782
return osErrorResult(CS_CONTEXT, ENOMEM, result);
783
if (!CSThread::attach(self))
784
return MSEngine::exceptionToResult(&self->myException, result);
793
int MSEngine::enterConnectionNoThd(CSThread **r_self, PBMSResultPtr result)
795
return enterConnection(current_thd, r_self, result, true);
799
void MSEngine::exitConnection()
801
THD *thd = (THD *) current_thd;
804
self = CSThread::getSelf();
805
if (self && self->pbms_api_owner)
810
CSThread::setSelf(NULL);
812
self = CSThread::getSelf();
813
CSThread::detach(self);
818
void MSEngine::closeConnection(THD* thd)
822
self = CSThread::getSelf();
823
if (self && self->pbms_api_owner)
827
if ((self = pbms_getMySelf(thd))) {
828
pbms_setMySelf(thd, NULL);
829
CSThread::setSelf(self);
830
CSThread::detach(self);
834
self = CSThread::getSelf();
835
CSThread::detach(self);