database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-29 08:21:15 +03:00
Code Activity

MCOL-66 - Make the DDL and DML parsers re-entrant.

Browse Source 
Serialize all DDL because the VVS can't handle modifying the same block simultaneously
Fix the CTRL+C logic in DML that caused COMMIT issues.
This commit is contained in:
David Hall
2016-07-20 10:13:28 -05:00
parent 3f16e1f4b1
commit 6d11ce030d
28 changed files with 1561 additions and 1027 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1066
dbcon/ddlpackage/ddl-scan.cpp
View File

File diff suppressed because it is too large Load Diff

85
dbcon/ddlpackage/ddl.l
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -29,16 +30,14 @@
#endif #endif
int lineno = 1; int lineno = 1;
void ddlerror(char *s); void ddlerror(yyscan_t yyscanner, char *s);
/* Handles to the buffer that the lexer uses internally */ static char* scanner_copy(char *str, yyscan_t yyscanner);
static YY_BUFFER_STATE scanbufhandle;
static char *scanbuf;
static char* scanner_copy (char *str);
%} %}
%option reentrant
%option bison-bridge
%option noyywrap %option noyywrap
%option nounput %option nounput
/* %option header-file="ddl-scan.h" */ /* %option header-file="ddl-scan.h" */
@ -83,16 +82,16 @@ CHARACTER {return IDB_CHAR;}
BIGINT {return BIGINT;} BIGINT {return BIGINT;}
CHECK {BEGIN(check1);return CHECK;} CHECK {BEGIN(check1);return CHECK;}
<check1>\( {BEGIN(check2); return '(';} <check1>\( {BEGIN(check2); return '(';}
<check2>[^)]*/\) {BEGIN(check1); ddllval.str = scanner_copy(ddltext); return CP_SEARCH_CONDITION_TEXT;} <check2>[^)]*/\) {BEGIN(check1); ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return CP_SEARCH_CONDITION_TEXT;}
<check1>\) {BEGIN(0); return ')';} <check1>\) {BEGIN(0); return ')';}
{quote} {BEGIN(inquote);return yytext[0];} {quote} {BEGIN(inquote);return yytext[0];}
<inquote>[^']*/' {BEGIN(endquote); ddllval.str = scanner_copy(ddltext); return SCONST;} <inquote>[^']*/' {BEGIN(endquote); ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return SCONST;}
<endquote>' {BEGIN(0); return yytext[0];} <endquote>' {BEGIN(0); return yytext[0];}
{integer} {ddllval.str = scanner_copy(ddltext); return ICONST;} {integer} {ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return ICONST;}
{decimal} {ddllval.str = scanner_copy(ddltext); return FCONST;} {decimal} {ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return FCONST;}
{real} {ddllval.str = scanner_copy(ddltext); return FCONST;} {real} {ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return FCONST;}
COMMENT {return COMMENT;} COMMENT {return COMMENT;}
COLUMN {return COLUMN;} COLUMN {return COLUMN;}
@ -101,7 +100,7 @@ CONSTRAINT {return CONSTRAINT;}
CONSTRAINTS {return CONSTRAINTS;} CONSTRAINTS {return CONSTRAINTS;}
CREATE {return CREATE;} CREATE {return CREATE;}
CURRENT_USER {return CURRENT_USER;} CURRENT_USER {return CURRENT_USER;}
DATE {ddllval.str=strdup("date"); return DATE;} DATE {ddlget_lval(yyscanner)->str=strdup("date"); return DATE;}
DATETIME {return DATETIME;} DATETIME {return DATETIME;}
DECIMAL {return DECIMAL;} DECIMAL {return DECIMAL;}
DEC {return DECIMAL;} DEC {return DECIMAL;}
@ -162,10 +161,10 @@ VARBINARY {return VARBINARY;}
/* ignore */ /* ignore */
} }
{identifier} {ddllval.str = scanner_copy(ddltext); return IDENT;} {identifier} {ddlget_lval(yyscanner)->str = scanner_copy(ddlget_text(yyscanner), yyscanner); return IDENT;}
{self} { {self} {
return ddltext[0]; return ddlget_text(yyscanner)[0];
} }
{grave_accent} { {grave_accent} {
@ -174,39 +173,44 @@ VARBINARY {return VARBINARY;}
%% %%
void ddlerror(char const *s) void ddlerror(yyscan_t yyscanner, char const *s)
{ {
printf("yyerror: %d: %s at %s\n", lineno, s, yytext); printf("yyerror: %d: %s at %s\n", lineno, s, ddlget_text(yyscanner));
} }
typedef std::vector<char*> valbuf_t; typedef std::vector<char*> valbuf_t;
static valbuf_t valbuf; #include <pthread.h>
#include "sqlparser.h"
using namespace ddlpackage;
/* /*
* Called before any actual parsing is done * Called before any actual parsing is done
*/ */
void scanner_init(const char *str) void scanner_init(const char* str, yyscan_t yyscanner)
{ {
size_t slen = strlen(str); size_t slen = strlen(str);
scan_data* pScanData = (scan_data*)ddlget_extra(yyscanner);
/* /*
* Might be left over after ereport() * Might be left over after ereport()
*/ */
struct yyguts_t * yyg = (struct yyguts_t*)yyscanner; // needed for macro YY_CURRENT_BUFFER
if (YY_CURRENT_BUFFER) if (YY_CURRENT_BUFFER)
yy_delete_buffer(YY_CURRENT_BUFFER); yy_delete_buffer(YY_CURRENT_BUFFER, yyscanner);
/* /*
* Make a scan buffer with special termination needed by flex. * Make a scan buffer with special termination needed by flex.
*/ */
scanbuf = (char *)malloc(slen + 2); pScanData->scanbuf = (char *)malloc(slen + 2);
memcpy(scanbuf, str, slen); memcpy(pScanData->scanbuf, str, slen);
scanbuf[slen] = scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR; pScanData->scanbuf[slen] = pScanData->scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR;
scanbufhandle = yy_scan_buffer(scanbuf, slen + 2); pScanData->scanbufhandle = (void*)yy_scan_buffer(pScanData->scanbuf, slen + 2, yyscanner);
std::cout << "scanner_init " << (uint64_t)pScanData->scanbufhandle << std::endl;
BEGIN(INITIAL); BEGIN(INITIAL);
valbuf.clear(); pScanData->valbuf.clear();
} }
@ -215,27 +219,28 @@ void scanner_init(const char *str)
*/ */
void scanner_finish(void) void scanner_finish(yyscan_t yyscanner)
{ {
char* str; char* str;
scan_data* pScanData = (scan_data*)ddlget_extra(yyscanner);
yy_delete_buffer(scanbufhandle); std::cout << "scanner_finish " << (uint64_t)pScanData->scanbufhandle << std::endl;
free(scanbuf); yy_delete_buffer((YY_BUFFER_STATE)pScanData->scanbufhandle, yyscanner);
free(pScanData->scanbuf);
unsigned int i; unsigned int i;
for(i=0; i<valbuf.size(); i++) { for(i=0; i<pScanData->valbuf.size(); i++) {
str = valbuf[i]; str = pScanData->valbuf[i];
if(str) { if(str) {
// std::cout << "valbuf:(" << str << ")" << std::endl; // std::cout << "valbuf:(" << str << ")" << std::endl;
free(valbuf[i]); free(pScanData->valbuf[i]);
} }
} }
valbuf.clear(); pScanData->valbuf.clear();
} }
char* scanner_copy (char *str) char* scanner_copy (char *str, yyscan_t yyscanner)
{ {
char* nv = strdup(str); char* nv = strdup(str);
if(nv) if(nv)
valbuf.push_back(nv); ((scan_data*)ddlget_extra(yyscanner))->valbuf.push_back(nv);
return nv; return nv;
} }

10
dbcon/ddlpackage/ddl.y
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -43,6 +44,7 @@
a more recent version of flex. At the time of this writing, our a more recent version of flex. At the time of this writing, our
development systems have: flex version 2.5.4 development systems have: flex version 2.5.4
MCOL-66 Modify to be a reentrant parser
*/ */
%{ %{
@ -60,13 +62,15 @@ using namespace ddlpackage;
/* The user is expect to pass a ParseTree* to grammar_init */ /* The user is expect to pass a ParseTree* to grammar_init */
static ParseTree* parseTree; static ParseTree* parseTree;
static std::string db_schema; static std::string db_schema;
int ddllex(); int ddllex(YYSTYPE* ddllval, void* yyscanner);
void ddlerror (char const *error); void ddlerror (void* yyscanner, char const *error);
char* copy_string(const char *str); char* copy_string(const char *str);
%} %}
%expect 15 %expect 15
%pure-parser
%lex-param {void * scanner}
%parse-param {void * scanner}
%debug %debug
/* Bison uses this to generate a C union definition. This is used to /* Bison uses this to generate a C union definition. This is used to

18
dbcon/ddlpackage/sqlparser.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -34,12 +35,13 @@
#include "ddl-gram.h" #include "ddl-gram.h"
#endif #endif
void scanner_finish(void); void scanner_finish(void* yyscanner);
void scanner_init(const char *str); void scanner_init(const char *str, void* yyscanner);
void grammar_init(ddlpackage::ParseTree *ptree, bool); void grammar_init(ddlpackage::ParseTree *ptree, bool);
int ddllex_init_extra(void* user_defined,void** yyscanner);
int ddllex_destroy(void* yyscanner);
int ddlparse(void* yyscanner);
void set_schema(std::string schema); void set_schema(std::string schema);
int ddlparse();
namespace ddlpackage { namespace ddlpackage {
using namespace std; using namespace std;
@ -62,9 +64,10 @@ namespace ddlpackage {
int SqlParser::Parse(const char* sqltext) int SqlParser::Parse(const char* sqltext)
{ {
scanner_init(sqltext); ddllex_init_extra(&scanData, &scanner);
scanner_init(sqltext, scanner);
grammar_init(&fParseTree, fDebug); grammar_init(&fParseTree, fDebug);
fStatus = ddlparse(); fStatus = ddlparse(scanner);
return fStatus; return fStatus;
} }
@ -86,7 +89,8 @@ namespace ddlpackage {
SqlParser::~SqlParser() SqlParser::~SqlParser()
{ {
scanner_finish(); // free scanner allocated memory scanner_finish(scanner); // free scanner allocated memory
ddllex_destroy(scanner);
} }

15
dbcon/ddlpackage/sqlparser.h
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -67,6 +68,18 @@ typedef SqlStatementList ParseTree;
@endverbatim @endverbatim
*/ */
/*
Instance specific data for use by the scanner.
*/
typedef std::vector<char*> valbuf_t;
struct scan_data
{
/* Handles to the buffer that the lexer uses internally */
char* scanbuf;
void* scanbufhandle; // This is a YY_BUFFER_STATE defined in ddl-scan.cpp
valbuf_t valbuf;
};
class SqlParser class SqlParser
{ {
@ -102,6 +115,8 @@ protected:
ParseTree fParseTree; ParseTree fParseTree;
int fStatus; ///< return from yyparse() stored here. int fStatus; ///< return from yyparse() stored here.
bool fDebug; ///< Turn on bison debugging. bool fDebug; ///< Turn on bison debugging.
void* scanner; // yyscan_t * needed for re-entrant flex scanner
scan_data scanData;
}; };

12
dbcon/ddlpackageproc/altertableprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -335,6 +336,8 @@ AlterTableProcessor::DDLResult AlterTableProcessor::processPackage(ddlpackage::A
if (i >= numTries) //error out if (i >= numTries) //error out
{ {
logging::Message::Args args; logging::Message::Args args;
string strOp("alter");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(sessionId); args.add(sessionId);
@ -644,7 +647,10 @@ void AlterTableProcessor::addColumn (uint32_t sessionID, execplan::CalpontSystem
aColumnList.push_back(columnDefPtr); aColumnList.push_back(columnDefPtr);
bool alterFlag = true; bool alterFlag = true;
if (inTableName.fSchema != CALPONT_SCHEMA) // MCOL-66 The DBRM can't handle concurrent DDL
boost::mutex::scoped_lock lk(dbrmMutex);
if (inTableName.fSchema != CALPONT_SCHEMA)
{ {
VERBOSE_INFO("Writing meta data to SYSCOL"); //send to WES to process VERBOSE_INFO("Writing meta data to SYSCOL"); //send to WES to process
bs.restart(); bs.restart();
@ -1061,6 +1067,10 @@ void AlterTableProcessor::dropColumn (uint32_t sessionID, execplan::CalpontSyste
OamCache * oamcache = OamCache::makeOamCache(); OamCache * oamcache = OamCache::makeOamCache();
boost::shared_ptr<std::map<int, int> > dbRootPMMap = oamcache->getDBRootToPMMap(); boost::shared_ptr<std::map<int, int> > dbRootPMMap = oamcache->getDBRootToPMMap();
pmNum = (*dbRootPMMap)[dbRoot]; pmNum = (*dbRootPMMap)[dbRoot];
// MCOL-66 The DBRM can't handle concurrent DDL
boost::mutex::scoped_lock lk(dbrmMutex);
try try
{ {
fWEClient->write(bytestream, (uint32_t)pmNum); fWEClient->write(bytestream, (uint32_t)pmNum);

34
dbcon/ddlpackageproc/createtableprocessor.cpp
View File

@ -249,7 +249,7 @@ keepGoing:
} }
fStartingColOID = fObjectIDManager.allocOIDs(numColumns+numDictCols+1); //include column, oids,dictionary oids and tableoid fStartingColOID = fObjectIDManager.allocOIDs(numColumns+numDictCols+1); //include column, oids,dictionary oids and tableoid
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "Create table allocOIDs got the stating oid " << fStartingColOID << endl; cout << fTxnid.id << " Create table allocOIDs got the starting oid " << fStartingColOID << endl;
#endif #endif
if (fStartingColOID < 0) if (fStartingColOID < 0)
{ {
@ -298,12 +298,14 @@ cout << "Create table allocOIDs got the stating oid " << fStartingColOID << endl
boost::shared_ptr<messageqcpp::ByteStream> bsIn; boost::shared_ptr<messageqcpp::ByteStream> bsIn;
boost::shared_ptr<std::map<int, int> > dbRootPMMap = oamcache->getDBRootToPMMap(); boost::shared_ptr<std::map<int, int> > dbRootPMMap = oamcache->getDBRootToPMMap();
pmNum = (*dbRootPMMap)[dbRoot]; pmNum = (*dbRootPMMap)[dbRoot];
// MCOL-66 The DBRM can't handle concurrent DDL
boost::mutex::scoped_lock lk(dbrmMutex);
try try
{ {
fWEClient->write(bytestream, (unsigned)pmNum);
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "create table sending We_SVR_WRITE_SYSTABLE to pm " << pmNum << endl; cout << fTxnid.id << " create table sending We_SVR_WRITE_SYSTABLE to pm " << pmNum << endl;
#endif #endif
fWEClient->write(bytestream, (unsigned)pmNum);
while (1) while (1)
{ {
bsIn.reset(new ByteStream()); bsIn.reset(new ByteStream());
@ -320,7 +322,7 @@ cout << "create table sending We_SVR_WRITE_SYSTABLE to pm " << pmNum << endl;
errorMsg.clear(); errorMsg.clear();
*bsIn >> errorMsg; *bsIn >> errorMsg;
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl; cout << fTxnid.id << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
#endif #endif
} }
break; break;
@ -330,7 +332,7 @@ cout << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
catch (runtime_error& ex) //write error catch (runtime_error& ex) //write error
{ {
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "create table got exception" << ex.what() << endl; cout << fTxnid.id << " create table got exception" << ex.what() << endl;
#endif #endif
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
errorMsg = ex.what(); errorMsg = ex.what();
@ -345,6 +347,9 @@ cout << "create table got unknown exception" << endl;
if (rc != 0) if (rc != 0)
{ {
#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
#endif
result.result =(ResultCode) rc; result.result =(ResultCode) rc;
Message::Args args; Message::Args args;
Message message(9); Message message(9);
@ -403,10 +408,10 @@ cout << "create table got unknown exception" << endl;
pmNum = (*dbRootPMMap)[dbRoot]; pmNum = (*dbRootPMMap)[dbRoot];
try try
{ {
fWEClient->write(bytestream, (uint32_t)pmNum);
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "create table sending We_SVR_WRITE_SYSTABLE to pm " << pmNum << endl; cout << fTxnid.id << " create table sending WE_SVR_WRITE_CREATE_SYSCOLUMN to pm " << pmNum << endl;
#endif #endif
fWEClient->write(bytestream, (uint32_t)pmNum);
while (1) while (1)
{ {
bsIn.reset(new ByteStream()); bsIn.reset(new ByteStream());
@ -423,7 +428,7 @@ cout << "create table sending We_SVR_WRITE_SYSTABLE to pm " << pmNum << endl;
errorMsg.clear(); errorMsg.clear();
*bsIn >> errorMsg; *bsIn >> errorMsg;
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl; cout << fTxnid.id << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
#endif #endif
} }
break; break;
@ -433,7 +438,7 @@ cout << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
catch (runtime_error& ex) //write error catch (runtime_error& ex) //write error
{ {
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "create table got exception" << ex.what() << endl; cout << fTxnid.id << " create table got exception" << ex.what() << endl;
#endif #endif
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
errorMsg = ex.what(); errorMsg = ex.what();
@ -442,12 +447,15 @@ cout << "create table got exception" << ex.what() << endl;
{ {
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
#ifdef IDB_DDL_DEBUG #ifdef IDB_DDL_DEBUG
cout << "create table got unknown exception" << endl; cout << fTxnid.id << " create table got unknown exception" << endl;
#endif #endif
} }
if (rc != 0) if (rc != 0)
{ {
#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Create table WE_SVR_WRITE_CREATE_SYSCOLUMN: " << errorMsg << endl;
#endif
result.result =(ResultCode) rc; result.result =(ResultCode) rc;
Message::Args args; Message::Args args;
Message message(9); Message message(9);
@ -572,6 +580,9 @@ cout << "create table got unknown exception" << endl;
pmNum = (*dbRootPMMap)[useDBRoot]; pmNum = (*dbRootPMMap)[useDBRoot];
try try
{ {
#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " create table sending WE_SVR_WRITE_CREATETABLEFILES to pm " << pmNum << endl;
#endif
fWEClient->write(bytestream, pmNum); fWEClient->write(bytestream, pmNum);
while (1) while (1)
{ {
@ -631,6 +642,9 @@ cout << "Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
if (rc != 0) if (rc != 0)
{ {
#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Create table We_SVR_WRITE_CREATETABLEFILES: " << errorMsg << endl;
#endif
rollBackTransaction( uniqueId, txnID, createTableStmt.fSessionID); //What to do with the error code rollBackTransaction( uniqueId, txnID, createTableStmt.fSessionID); //What to do with the error code
fSessionManager.rolledback(txnID); fSessionManager.rolledback(txnID);
} }

2
dbcon/ddlpackageproc/ddlpackageprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -80,6 +81,7 @@ using namespace ddlpackage;
namespace ddlpackageprocessor namespace ddlpackageprocessor
{ {
boost::mutex DDLPackageProcessor::dbrmMutex;
DDLPackageProcessor::~DDLPackageProcessor() DDLPackageProcessor::~DDLPackageProcessor()
{ {

3
dbcon/ddlpackageproc/ddlpackageprocessor.h
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -802,6 +803,8 @@ protected:
int rollBackTransaction(uint64_t uniqueId, BRM::TxnID txnID, uint32_t sessionID); int rollBackTransaction(uint64_t uniqueId, BRM::TxnID txnID, uint32_t sessionID);
int commitTransaction(uint64_t uniqueId, BRM::TxnID txnID); int commitTransaction(uint64_t uniqueId, BRM::TxnID txnID);
// MCOL-66 The DBRM can't handle concurrent DDL
static boost::mutex dbrmMutex;
private: private:
/** @brief clean beginning and ending glitches and spaces from string /** @brief clean beginning and ending glitches and spaces from string
* *

3
dbcon/ddlpackageproc/droppartitionprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -196,6 +197,8 @@ namespace ddlpackageprocessor
{ {
result.result = DROP_ERROR; result.result = DROP_ERROR;
logging::Message::Args args; logging::Message::Args args;
string strOp("drop partition");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add((uint64_t)sessionID); args.add((uint64_t)sessionID);

78
dbcon/ddlpackageproc/droptableprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -123,7 +124,11 @@ DropTableProcessor::DDLResult DropTableProcessor::processPackage(ddlpackage::Dro
uint64_t tableLockId = 0; uint64_t tableLockId = 0;
OamCache* oamcache = OamCache::makeOamCache(); OamCache* oamcache = OamCache::makeOamCache();
std::vector<int> moduleIds = oamcache->getModuleIds(); std::vector<int> moduleIds = oamcache->getModuleIds();
try
// MCOL-66 The DBRM can't handle concurrent DDL
boost::mutex::scoped_lock lk(dbrmMutex);
try
{ {
//check table lock //check table lock
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(dropTableStmt.fSessionID); boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(dropTableStmt.fSessionID);
@ -197,6 +202,8 @@ DropTableProcessor::DDLResult DropTableProcessor::processPackage(ddlpackage::Dro
if (i >= numTries) //error out if (i >= numTries) //error out
{ {
Message::Args args; Message::Args args;
string strOp("drop table");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(sessionId); args.add(sessionId);
@ -238,9 +245,9 @@ DropTableProcessor::DDLResult DropTableProcessor::processPackage(ddlpackage::Dro
//get a unique number //get a unique number
VERBOSE_INFO("Removing the SYSTABLE meta data"); VERBOSE_INFO("Removing the SYSTABLE meta data");
#ifdef IDB_DDL_DEBUG //#ifdef IDB_DDL_DEBUG
cout << "Removing the SYSTABLEs meta data" << endl; cout << fTxnid.id << " Removing the SYSTABLEs meta data" << endl;
#endif //#endif
bytestream << (ByteStream::byte)WE_SVR_DELETE_SYSTABLE; bytestream << (ByteStream::byte)WE_SVR_DELETE_SYSTABLE;
bytestream << uniqueId; bytestream << uniqueId;
bytestream << (uint32_t) dropTableStmt.fSessionID; bytestream << (uint32_t) dropTableStmt.fSessionID;
@ -271,11 +278,11 @@ cout << "Removing the SYSTABLEs meta data" << endl;
pmNum = (*dbRootPMMap)[dbRoot]; pmNum = (*dbRootPMMap)[dbRoot];
try try
{ {
// #ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Drop table sending WE_SVR_DELETE_SYSTABLES to pm " << pmNum << endl;
//#endif
//cout << "deleting systable entries with txnid " << txnID.id << endl; //cout << "deleting systable entries with txnid " << txnID.id << endl;
fWEClient->write(bytestream, (uint32_t)pmNum); fWEClient->write(bytestream, (uint32_t)pmNum);
#ifdef IDB_DDL_DEBUG
cout << "Drop table sending WE_SVR_DELETE_SYSTABLES to pm " << pmNum << endl;
#endif
while (1) while (1)
{ {
bsIn.reset(new ByteStream()); bsIn.reset(new ByteStream());
@ -297,22 +304,23 @@ cout << "Drop table sending WE_SVR_DELETE_SYSTABLES to pm " << pmNum << endl;
} }
catch (runtime_error& ex) //write error catch (runtime_error& ex) //write error
{ {
#ifdef IDB_DDL_DEBUG // #ifdef IDB_DDL_DEBUG
cout << "Drop table got exception" << endl; cout << fTxnid.id << " Drop table got exception" << endl;
#endif // #endif
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
errorMsg = ex.what(); errorMsg = ex.what();
} }
catch (...) catch (...)
{ {
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
#ifdef IDB_DDL_DEBUG //#ifdef IDB_DDL_DEBUG
cout << "Drop table got unknown exception" << endl; cout << fTxnid.id << " Drop table got unknown exception" << endl;
#endif //#endif
} }
if (rc != 0) if (rc != 0)
{ {
cout << fTxnid.id << " Error in dropping table from systables(" << (int)rc << ") " << errorMsg.c_str() << endl;
Message::Args args; Message::Args args;
Message message(9); Message message(9);
args.add("Error in dropping table from systables."); args.add("Error in dropping table from systables.");
@ -355,11 +363,10 @@ cout << "Drop table got unknown exception" << endl;
pmNum = (*dbRootPMMap)[dbRoot]; pmNum = (*dbRootPMMap)[dbRoot];
try try
{ {
//cout << "deleting systable entries with txnid " << txnID.id << endl; //#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Drop table sending WE_SVR_DELETE_SYSCOLUMN to pm " << pmNum << endl;
//#endif
fWEClient->write(bytestream, (unsigned)pmNum); fWEClient->write(bytestream, (unsigned)pmNum);
#ifdef IDB_DDL_DEBUG
cout << "Drop table sending WE_SVR_DELETE_SYSTABLES to pm " << pmNum << endl;
#endif
while (1) while (1)
{ {
bsIn.reset(new ByteStream()); bsIn.reset(new ByteStream());
@ -381,25 +388,26 @@ cout << "Drop table sending WE_SVR_DELETE_SYSTABLES to pm " << pmNum << endl;
} }
catch (runtime_error& ex) //write error catch (runtime_error& ex) //write error
{ {
#ifdef IDB_DDL_DEBUG //#ifdef IDB_DDL_DEBUG
cout << "Drop table got exception" << endl; cout << fTxnid.id << " Drop table got exception" << endl;
#endif //#endif
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
errorMsg = ex.what(); errorMsg = ex.what();
} }
catch (...) catch (...)
{ {
rc = NETWORK_ERROR; rc = NETWORK_ERROR;
#ifdef IDB_DDL_DEBUG // #ifdef IDB_DDL_DEBUG
cout << "Drop table got unknown exception" << endl; cout << fTxnid.id << " Drop table got unknown exception" << endl;
#endif //#endif
} }
if (rc != 0) if (rc != 0)
{ {
cout << fTxnid.id << " Error in dropping column from systables(" << (int)rc << ") " << errorMsg.c_str() << endl;
Message::Args args; Message::Args args;
Message message(9); Message message(9);
args.add("Error in dropping table from systables."); args.add("Error in dropping column from systables.");
args.add(errorMsg); args.add(errorMsg);
message.format(args); message.format(args);
result.result = (ResultCode)rc; result.result = (ResultCode)rc;
@ -412,11 +420,16 @@ cout << "Drop table got unknown exception" << endl;
} }
rc = commitTransaction(uniqueId, txnID); rc = commitTransaction(uniqueId, txnID);
//cout << "commiting transaction " << txnID.id << " and valid is " << txnID.valid << endl;
if (rc != 0) if (rc != 0)
{
cout << txnID.id << " rolledback transaction " << " and valid is " << txnID.valid << endl;
fSessionManager.rolledback(txnID); fSessionManager.rolledback(txnID);
}
else else
{
cout << txnID.id << " commiting transaction " << txnID.id << " and valid is " << txnID.valid << endl;
fSessionManager.committed(txnID); fSessionManager.committed(txnID);
}
if (rc != 0) if (rc != 0)
{ {
@ -527,9 +540,9 @@ cout << "Drop table got unknown exception" << endl;
{ {
bytestream << (uint32_t) oidList[i]; bytestream << (uint32_t) oidList[i];
} }
#ifdef IDB_DDL_DEBUG //#ifdef IDB_DDL_DEBUG
cout << "Drop table removing column files" << endl; cout << fTxnid.id << " Drop table removing column files" << endl;
#endif //#endif
uint32_t msgRecived = 0; uint32_t msgRecived = 0;
try { try {
fWEClient->write_to_all(bytestream); fWEClient->write_to_all(bytestream);
@ -592,6 +605,9 @@ cout << "Drop table removing column files" << endl;
//Flush primProc cache //Flush primProc cache
rc = cacheutils::flushOIDsFromCache( oidList ); rc = cacheutils::flushOIDsFromCache( oidList );
//Delete extents from extent map //Delete extents from extent map
//#ifdef IDB_DDL_DEBUG
cout << fTxnid.id << " Drop table deleteOIDs" << endl;
//#endif
rc = fDbrm->deleteOIDs(oidList); rc = fDbrm->deleteOIDs(oidList);
if (rc != 0) if (rc != 0)
@ -879,7 +895,11 @@ TruncTableProcessor::DDLResult TruncTableProcessor::processPackage(ddlpackage::T
ByteStream bytestream; ByteStream bytestream;
ByteStream::byte tmp8; ByteStream::byte tmp8;
try {
// MCOL-66 The DBRM can't handle concurrent DDL
boost::mutex::scoped_lock lk(dbrmMutex);
try {
//Disable extents first //Disable extents first
int rc1 = fDbrm->markAllPartitionForDeletion( allOidList); int rc1 = fDbrm->markAllPartitionForDeletion( allOidList);
if (rc1 != 0) if (rc1 != 0)

3
dbcon/ddlpackageproc/restorepartitionprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -160,6 +161,8 @@ namespace ddlpackageprocessor
{ {
result.result = DROP_ERROR; result.result = DROP_ERROR;
logging::Message::Args args; logging::Message::Args args;
string strOp("restore partition");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add((uint64_t)sessionID); args.add((uint64_t)sessionID);

1029
dbcon/dmlpackage/dml-scan.cpp
View File

File diff suppressed because it is too large Load Diff

56
dbcon/dmlpackage/dml.l
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -44,25 +45,23 @@ extern "C" int _isatty(int);
#define yyerror dml_yyerror*/ #define yyerror dml_yyerror*/
using namespace dmlpackage; using namespace dmlpackage;
void dmlerror(char const *s); void dmlerror(yyscan_t yyscanner, char const *s);
namespace dmlpackage { namespace dmlpackage {
int lineno = 1; int lineno = 1;
/* Handles to the buffer that the lexer uses internally */ static char* scanner_copy (char *str, yyscan_t yyscanner);
static YY_BUFFER_STATE scanbufhandle;
static char *scanbuf;
static char* scanner_copy (char *str);
/* macro to save the text and return a token */ /* macro to save the text and return a token */
#define TOK(name) { dmllval.strval = scanner_copy(dmltext); return name; } #define TOK(name) { dmlget_lval(yyscanner)->strval = scanner_copy(dmlget_text(yyscanner), yyscanner); return name; }
} }
%} %}
%option reentrant
%option bison-bridge
%option noyywrap %option noyywrap
%option nounput %option nounput
%x inquote %x inquote
@ -199,9 +198,9 @@ WORK TOK(WORK)
%% %%
using namespace dmlpackage; using namespace dmlpackage;
void dmlerror(char const *s) void dmlerror(yyscan_t yyscanner, char const *s)
{ {
printf("yyerror: %d: %s at %s\n", lineno, s, yytext); printf("yyerror: %d: %s at %s\n", lineno, s, dmlget_text(yyscanner));
} }
namespace dmlpackage { namespace dmlpackage {
@ -216,28 +215,30 @@ valbuf_t get_valbuffer(void)
/* /*
* Called before any actual parsing is done * Called before any actual parsing is done
*/ */
void scanner_init(const char *str) void scanner_init(const char *str, yyscan_t yyscanner)
{ {
size_t slen = strlen(str); size_t slen = strlen(str);
scan_data* pScanData = (scan_data*)dmlget_extra(yyscanner);
/* /*
* Might be left over after ereport() * Might be left over after ereport()
*/ */
struct yyguts_t * yyg = (struct yyguts_t*)yyscanner; // needed for macro YY_CURRENT_BUFFER
if (YY_CURRENT_BUFFER) if (YY_CURRENT_BUFFER)
yy_delete_buffer(YY_CURRENT_BUFFER); yy_delete_buffer(YY_CURRENT_BUFFER, yyscanner);
/* /*
* Make a scan buffer with special termination needed by flex. * Make a scan buffer with special termination needed by flex.
*/ */
scanbuf = (char *)malloc(slen + 2); pScanData->scanbuf = (char *)malloc(slen + 2);
memcpy(scanbuf, str, slen); memcpy(pScanData->scanbuf, str, slen);
scanbuf[slen] = scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR; pScanData->scanbuf[slen] = pScanData->scanbuf[slen + 1] = YY_END_OF_BUFFER_CHAR;
scanbufhandle = yy_scan_buffer(scanbuf, slen + 2); pScanData->scanbufhandle = yy_scan_buffer(pScanData->scanbuf, slen + 2, yyscanner);
BEGIN(INITIAL); BEGIN(INITIAL);
valbuf.clear(); pScanData->valbuf.clear();
} }
@ -246,28 +247,29 @@ void scanner_init(const char *str)
*/ */
void scanner_finish(void) void scanner_finish(yyscan_t yyscanner)
{ {
char* str; char* str;
scan_data* pScanData = (scan_data*)dmlget_extra(yyscanner);
yy_delete_buffer(scanbufhandle); yy_delete_buffer((YY_BUFFER_STATE)pScanData->scanbufhandle, yyscanner);
free(scanbuf); free(pScanData->scanbuf);
unsigned int i; unsigned int i;
for(i=0; i<valbuf.size(); i++) { for(i=0; i<pScanData->valbuf.size(); i++) {
str = valbuf[i]; str = pScanData->valbuf[i];
if(str) { if(str) {
//std::cout << "valbuf:(" << str << ")" << std::endl; //std::cout << "valbuf:(" << str << ")" << std::endl;
free(valbuf[i]); free(pScanData->valbuf[i]);
} }
} }
valbuf.clear(); pScanData->valbuf.clear();
} }
char* scanner_copy (char *str) char* scanner_copy (char *str, yyscan_t yyscanner)
{ {
char* nv = strdup(str); char* nv = strdup(str);
if(nv) if(nv)
valbuf.push_back(nv); ((scan_data*)dmlget_extra(yyscanner))->valbuf.push_back(nv);
return nv; return nv;
} }

12
dbcon/dmlpackage/dml.y
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -55,6 +56,7 @@
a more recent version of flex. At the time of this writing, our a more recent version of flex. At the time of this writing, our
development systems have: flex version 2.5.4 development systems have: flex version 2.5.4
MCOL-66 Modify to be a reentrant parser
*/ */
%{ %{
#include <string.h> #include <string.h>
@ -72,9 +74,9 @@
using namespace std; using namespace std;
using namespace dmlpackage; using namespace dmlpackage;
int dmllex(); int dmllex(YYSTYPE* dmllval, void* yyscanner);
void dmlerror (char const *error); void dmlerror (void* yyscanner, char const *error);
namespace dmlpackage { namespace dmlpackage {
@ -88,8 +90,12 @@ char* copy_string(const char *str);
} }
%} %}
%pure-parser
%lex-param {void * scanner}
%parse-param {void * scanner}
%debug %debug
/* symbolic tokens */
/* symbolic tokens */
%union { %union {
int intval; int intval;

17
dbcon/dmlpackage/dmlparser.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -37,14 +38,16 @@
#include <stdio.h> #include <stdio.h>
#include <string.h> #include <string.h>
int dmlparse(); int dmllex_init_extra(void* user_defined, void** yyscanner);
int dmllex_destroy(void* yyscanner);
int dmlparse(void* yyscanner);
namespace dmlpackage namespace dmlpackage
{ {
using namespace std; using namespace std;
void scanner_finish(void); void scanner_finish(void* yyscanner);
void scanner_init(const char *str); void scanner_init(const char *str, void* yyscanner);
void grammar_init(dmlpackage::ParseTree* _ptree, bool); void grammar_init(dmlpackage::ParseTree* _ptree, bool);
valbuf_t get_valbuffer(void); valbuf_t get_valbuffer(void);
@ -57,7 +60,8 @@ namespace dmlpackage
DMLParser::~DMLParser() DMLParser::~DMLParser()
{ {
scanner_finish(); scanner_finish(scanner);
dmllex_destroy(scanner);
} }
void DMLParser::setDebug(bool debug) void DMLParser::setDebug(bool debug)
@ -67,9 +71,10 @@ namespace dmlpackage
int DMLParser::parse(const char* dmltext) int DMLParser::parse(const char* dmltext)
{ {
scanner_init(dmltext); dmllex_init_extra(&scanData, &scanner);
scanner_init(dmltext, scanner);
grammar_init(&fParseTree, fDebug); grammar_init(&fParseTree, fDebug);
fStatus = dmlparse(); fStatus = dmlparse(scanner);
if (fStatus == 0) if (fStatus == 0)
{ {
char* str; char* str;

20
dbcon/dmlpackage/dmlparser.h
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -35,9 +36,20 @@ namespace dmlpackage
typedef SqlStatementList ParseTree; typedef SqlStatementList ParseTree;
/** @brief BISON parser wrapper class // instance data for the parser
*/ typedef std::vector<char*> valbuf_t;
class DMLParser
struct scan_data
{
/* Handles to the buffer that the lexer uses internally */
char* scanbuf;
void* scanbufhandle; // This is a YY_BUFFER_STATE defined in ddl-scan.cpp
valbuf_t valbuf;
};
/** @brief BISON parser wrapper class
*/
class DMLParser
{ {
public: public:
/** @brief ctor /** @brief ctor
@ -73,6 +85,8 @@ namespace dmlpackage
ParseTree fParseTree; ParseTree fParseTree;
int fStatus; int fStatus;
bool fDebug; bool fDebug;
void* scanner; // yyscan_t * needed for re-entrant flex scanner
scan_data scanData;
private: private:

3
dbcon/dmlpackageproc/deletepackageprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -190,6 +191,8 @@ namespace dmlpackageprocessor
{ {
result.result = DELETE_ERROR; result.result = DELETE_ERROR;
logging::Message::Args args; logging::Message::Args args;
string strOp("delete");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(sessionId); args.add(sessionId);

3
dbcon/dmlpackageproc/insertpackageprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -197,6 +198,8 @@ namespace dmlpackageprocessor
{ {
result.result = INSERT_ERROR; result.result = INSERT_ERROR;
logging::Message::Args args; logging::Message::Args args;
string strOp("insert");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(sessionId); args.add(sessionId);

3
dbcon/dmlpackageproc/updatepackageprocessor.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -214,6 +215,8 @@ UpdatePackageProcessor::processPackage(dmlpackage::CalpontDMLPackage& cpackage)
{ {
result.result = UPDATE_ERROR; result.result = UPDATE_ERROR;
logging::Message::Args args; logging::Message::Args args;
string strOp("update");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(sessionId); args.add(sessionId);

6
dbcon/mysql/mysql.vpj
View File

@ -245,10 +245,15 @@
<F N="../../../mariadb/sql/group_by_handler.h"/> <F N="../../../mariadb/sql/group_by_handler.h"/>
<F N="../../../mariadb/sql/gstream.h"/> <F N="../../../mariadb/sql/gstream.h"/>
<F N="ha_calpont.h"/> <F N="ha_calpont.h"/>
<F N="ha_calpont_impl.h"/>
<F N="ha_calpont_impl_if.h"/>
<F N="../../../mariadb/sql/ha_partition.h"/> <F N="../../../mariadb/sql/ha_partition.h"/>
<F N="ha_subquery.h"/>
<F N="ha_view.h"/>
<F N="../../../mariadb/sql/handler.h"/> <F N="../../../mariadb/sql/handler.h"/>
<F N="../../../mariadb/sql/hash_filo.h"/> <F N="../../../mariadb/sql/hash_filo.h"/>
<F N="../../../mariadb/sql/hostname.h"/> <F N="../../../mariadb/sql/hostname.h"/>
<F N="idb_mysql.h"/>
<F N="../../../mariadb/sql/init.h"/> <F N="../../../mariadb/sql/init.h"/>
<F N="../../../mariadb/sql/innodb_priv.h"/> <F N="../../../mariadb/sql/innodb_priv.h"/>
<F N="../../../mariadb/sql/item.h"/> <F N="../../../mariadb/sql/item.h"/>
@ -312,6 +317,7 @@
<F N="../../../mariadb/sql/scheduler.h"/> <F N="../../../mariadb/sql/scheduler.h"/>
<F N="../../../mariadb/sql/set_var.h"/> <F N="../../../mariadb/sql/set_var.h"/>
<F N="../../../mariadb/sql/slave.h"/> <F N="../../../mariadb/sql/slave.h"/>
<F N="sm.h"/>
<F N="../../../mariadb/sql/sp.h"/> <F N="../../../mariadb/sql/sp.h"/>
<F N="../../../mariadb/sql/sp_cache.h"/> <F N="../../../mariadb/sql/sp_cache.h"/>
<F N="../../../mariadb/sql/sp_head.h"/> <F N="../../../mariadb/sql/sp_head.h"/>

3
ddlproc/ddlproc.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -117,7 +118,7 @@ int main(int argc, char* argv[])
sigaction(SIGPIPE, &ign, 0); sigaction(SIGPIPE, &ign, 0);
#endif #endif
ddlprocessor::DDLProcessor ddlprocessor(5, 10); ddlprocessor::DDLProcessor ddlprocessor(1, 20);
{ {
Oam oam; Oam oam;

2
dmlproc/batchinsertprocessor.cpp
View File

@ -152,6 +152,8 @@ uint64_t BatchInsertProc::grabTableLock(int32_t sessionId)
if (i >= numTries) //error out if (i >= numTries) //error out
{ {
logging::Message::Args args; logging::Message::Args args;
string strOp("batch insert");
args.add(strOp);
args.add(processName); args.add(processName);
args.add((uint64_t)processID); args.add((uint64_t)processID);
args.add(tmpSessionId); args.add(tmpSessionId);

1
dmlproc/batchinsertprocessor.h
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License

80
dmlproc/dmlprocessor.cpp
View File

@ -1,5 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; version 2 of as published by the Free Software Foundation; version 2 of
@ -310,7 +310,6 @@ void PackageHandler::run()
try try
{ {
switch( fPackageType ) switch( fPackageType )
{ {
case dmlpackage::DML_INSERT: case dmlpackage::DML_INSERT:
@ -785,29 +784,53 @@ void PackageHandler::run()
ml.logWarningMessage( result.message ); ml.logWarningMessage( result.message );
} }
// send back the results
messageqcpp::ByteStream results;
messageqcpp::ByteStream::octbyte rowCount = result.rowCount;
messageqcpp::ByteStream::byte retval = result.result;
results << retval;
results << rowCount;
results << result.message.msg();
results << result.tableLockInfo; // ? connector does not get
// query stats
results << result.queryStats;
results << result.extendedStats;
results << result.miniStats;
result.stats.serialize(results);
fIos.write(results);
//Bug 5226. dmlprocessor thread will close the socket to mysqld.
//if (stmt == "CLEANUP")
// fIos.close();
} }
catch(std::exception& e)
{
cout << "dmlprocessor.cpp PackageHandler::run() package type("
<< fPackageType << ") exception: " << e.what() << endl;
logging::LoggingID lid(21);
logging::MessageLog ml(lid);
logging::Message::Args args;
logging::Message message(1);
args.add("dmlprocessor.cpp PackageHandler::run() package type");
args.add((uint64_t)fPackageType);
args.add(e.what());
message.format(args);
ml.logErrorMessage(message);
result.result=DMLPackageProcessor::COMMAND_ERROR;
result.message = message;
}
catch(...) catch(...)
{ {
fIos.close(); logging::LoggingID lid(21);
logging::MessageLog ml(lid);
logging::Message::Args args;
logging::Message message(1);
args.add("dmlprocessor.cpp PackageHandler::run() ... exception package type");
args.add((uint64_t)fPackageType);
message.format(args);
ml.logErrorMessage(message);
result.result=DMLPackageProcessor::COMMAND_ERROR;
result.message = message;
} }
// send back the results
messageqcpp::ByteStream results;
messageqcpp::ByteStream::octbyte rowCount = result.rowCount;
messageqcpp::ByteStream::byte retval = result.result;
results << retval;
results << rowCount;
results << result.message.msg();
results << result.tableLockInfo; // ? connector does not get
// query stats
results << result.queryStats;
results << result.extendedStats;
results << result.miniStats;
result.stats.serialize(results);
fIos.write(results);
//Bug 5226. dmlprocessor thread will close the socket to mysqld.
//if (stmt == "CLEANUP")
// fIos.close();
} }
void PackageHandler::rollbackPending() void PackageHandler::rollbackPending()
@ -924,15 +947,15 @@ void DMLProcessor::operator()()
bs1.reset(new messageqcpp::ByteStream(fIos.read())); bs1.reset(new messageqcpp::ByteStream(fIos.read()));
//cout << "received from mysql socket " << fIos.getSockID() << endl; //cout << "received from mysql socket " << fIos.getSockID() << endl;
} }
catch (std::exception&) catch (std::exception& ex)
{ {
//This is an I/O error from InetStreamSocket::read(), just close and move on... //This is an I/O error from InetStreamSocket::read(), just close and move on...
//cout << "runtime error during read on " << fIos.getSockID() << " " << ex.what() << endl; cout << "runtime error during read on " << fIos.getSockID() << " " << ex.what() << endl;
bs1->reset(); bs1->reset();
} }
catch (...) catch (...)
{ {
//cout << "... error during read " << fIos.getSockID() << endl; cout << "... error during read " << fIos.getSockID() << endl;
// all this throw does is cause this thread to silently go away. I doubt this is the right // all this throw does is cause this thread to silently go away. I doubt this is the right
// thing to do... // thing to do...
throw; throw;
@ -940,7 +963,7 @@ void DMLProcessor::operator()()
if (!bs1 || bs1->length() == 0) if (!bs1 || bs1->length() == 0)
{ {
//cout << "Read 0 bytes. Closing connection " << fIos.getSockID() << endl; cout << "Read 0 bytes. Closing connection " << fIos.getSockID() << endl;
fIos.close(); fIos.close();
break; break;
} }
@ -1069,8 +1092,11 @@ void DMLProcessor::operator()()
{ {
if (packageType == dmlpackage::DML_COMMAND) if (packageType == dmlpackage::DML_COMMAND)
{ {
// MCOL-66 It's possible for a commit or rollback to get here if
// the timing is just right. Don't destroy its data
messageqcpp::ByteStream bsctrlc(bs1);
dmlpackage::CommandDMLPackage commandPkg; dmlpackage::CommandDMLPackage commandPkg;
commandPkg.read(*(bs1.get())); commandPkg.read(bsctrlc);
std::string stmt = commandPkg.get_DMLStatement(); std::string stmt = commandPkg.get_DMLStatement();
boost::algorithm::to_upper(stmt); boost::algorithm::to_upper(stmt);
trim(stmt); trim(stmt);
@ -1288,7 +1314,7 @@ void DMLProcessor::operator()()
{ {
if (packageType != dmlpackage::DML_COMMAND) if (packageType != dmlpackage::DML_COMMAND)
{ {
txnid = sessionManager.getTxnID(sessionID); txnid = sessionManager.getTxnID(sessionID);
if ( !txnid.valid ) if ( !txnid.valid )
{ {
txnid = sessionManager.newTxnID(sessionID, true); txnid = sessionManager.newTxnID(sessionID, true);

15
writeengine/server/we_ddlcommandproc.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -159,8 +160,8 @@ uint8_t WE_DDLCommandProc::writeSystable(ByteStream& bs, std::string &err)
getColumnsForTable(sessionID, tableName.schema,tableName.table, columns); getColumnsForTable(sessionID, tableName.schema,tableName.table, columns);
column_iterator = columns.begin(); column_iterator = columns.begin();
std::string tmpStr(""); std::string tmpStr("");
while (column_iterator != columns.end()) while (column_iterator != columns.end())
{ {
@ -278,13 +279,17 @@ uint8_t WE_DDLCommandProc::writeSystable(ByteStream& bs, std::string &err)
++column_iterator; ++column_iterator;
} }
//fWriteEngine.setDebugLevel(WriteEngine::DEBUG_3); //fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
fWEWrapper.setTransId(txnID); fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true); fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(false); fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID); fWEWrapper.startTransaction(txnID);
if (0 != colStructs.size()) if (0 != colStructs.size())
{ {
// MCOL-66 The DBRM can't handle concurrent transactions to sys tables
// TODO: This may be redundant
static boost::mutex dbrmMutex;
boost::mutex::scoped_lock lk(dbrmMutex);
error = fWEWrapper.insertColumnRec_SYS(txnID, colStructs, colValuesList, error = fWEWrapper.insertColumnRec_SYS(txnID, colStructs, colValuesList,
dctnryStructList, dctnryValueList, SYSCOLUMN_BASE); dctnryStructList, dctnryValueList, SYSCOLUMN_BASE);
@ -683,7 +688,7 @@ uint8_t WE_DDLCommandProc::writeCreateSyscolumn(ByteStream& bs, std::string &err
colValuesList.push_back(colList[n]); colValuesList.push_back(colList[n]);
dctnryValueList.push_back(dctColList[n]); dctnryValueList.push_back(dctColList[n]);
} }
//fWriteEngine.setDebugLevel(WriteEngine::DEBUG_3); //fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
error = fWEWrapper.insertColumnRec_SYS(txnID, colStructs, colValuesList, error = fWEWrapper.insertColumnRec_SYS(txnID, colStructs, colValuesList,
dctnryStructList, dctnryValueList, SYSCOLUMN_BASE); dctnryStructList, dctnryValueList, SYSCOLUMN_BASE);
@ -1042,7 +1047,7 @@ uint8_t WE_DDLCommandProc::writeSyscolumn(ByteStream& bs, std::string & err)
colValuesList.push_back(colList[n]); colValuesList.push_back(colList[n]);
dctnryValueList.push_back(dctColList[n]); dctnryValueList.push_back(dctColList[n]);
} }
//fWriteEngine.setDebugLevel(WriteEngine::DEBUG_3); //fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
fWEWrapper.setTransId(txnID); fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true); fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(false); fWEWrapper.setBulkFlag(false);

5
writeengine/server/we_dmlcommandproc.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -3068,10 +3069,10 @@ uint8_t WE_DMLCommandProc::processBulkRollback(messageqcpp::ByteStream& bs,
// but it shouldn't hurt to keep in here. // but it shouldn't hurt to keep in here.
std::cout << "processBulkRollback"; std::cout << "processBulkRollback";
bs >> tableLockID; bs >> tableLockID;
std::cout << ": tableLock-"<< tableLockID; //std::cout << ": tableLock-"<< tableLockID;
bs >> tableOID; bs >> tableOID;
std::cout << "; tableOID-" << tableOID; //std::cout << "; tableOID-" << tableOID;
bs >> tableName; bs >> tableName;
if (tableName.length() == 0) if (tableName.length() == 0)

4
writeengine/wrapper/writeengine.cpp
View File

@ -1,4 +1,5 @@
/* Copyright (C) 2014 InfiniDB, Inc. /* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 MariaDB Corporation
This program is free software; you can redistribute it and/or This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License modify it under the terms of the GNU General Public License
@ -1955,6 +1956,9 @@ timer.stop("tokenize");
if (rc == NO_ERROR) if (rc == NO_ERROR)
{ {
// MCOL-66 The DBRM can't handle concurrent transactions to sys tables
static boost::mutex dbrmMutex;
boost::mutex::scoped_lock lk(dbrmMutex);
if (newExtent) if (newExtent)
{ {
rc = writeColumnRec(txnid, colStructList, colOldValueList, rowIdArray, newColStructList, colNewValueList, tableOid, false); // @bug 5572 HDFS tmp file rc = writeColumnRec(txnid, colStructList, colOldValueList, rowIdArray, newColStructList, colNewValueList, tableOid, false); // @bug 5572 HDFS tmp file