database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-30 19:23:07 +03:00
Code Activity

fix(logging): Add setddldebuglevel command (#3312) (#3344)

Browse Source
This commit is contained in:
Denis Khalikov
2024-11-10 00:38:19 +03:00
committed by GitHub
parent 11136b3545
commit 4cba8301d5
10 changed files with 345 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
dbcon/ddlpackage/CMakeLists.txt
View File

@ -32,6 +32,7 @@ ADD_LIBRARY(ddlpackage SHARED
markpartition.cpp markpartition.cpp
restorepartition.cpp restorepartition.cpp
droppartition.cpp droppartition.cpp
debugstatement.cpp
${BISON_ddl_gram_OUTPUTS} ${BISON_ddl_gram_OUTPUTS}
${FLEX_ddl_scan_OUTPUTS} ${FLEX_ddl_scan_OUTPUTS}
) )

29
dbcon/ddlpackage/ddlpkg.h
View File

@ -332,7 +332,8 @@ enum DDL_SERIAL_TYPE
DDL_TRUNC_TABLE_STATEMENT, DDL_TRUNC_TABLE_STATEMENT,
DDL_MARK_PARTITION_STATEMENT, DDL_MARK_PARTITION_STATEMENT,
DDL_RESTORE_PARTITION_STATEMENT, DDL_RESTORE_PARTITION_STATEMENT,
DDL_DROP_PARTITION_STATEMENT DDL_DROP_PARTITION_STATEMENT,
DDL_DEBUG_STATEMENT
}; };
/** @brief An abstract base for TableDef, ColumnDef, ... /** @brief An abstract base for TableDef, ColumnDef, ...
@ -1319,8 +1320,10 @@ struct AlterTableStatement : public SqlStatement
QualifiedName* fTableName; QualifiedName* fTableName;
AlterTableActionList fActions; AlterTableActionList fActions;
private: private:
long fTimeZone; long fTimeZone;
public: public:
}; };
@ -1441,6 +1444,30 @@ struct DropTableStatement : public SqlStatement
bool fCascade; bool fCascade;
}; };
/** @brief DebugStatement
*/
struct DebugDDLStatement : public SqlStatement
{
/** @brief Deserialize from ByteStream */
EXPORT virtual int unserialize(messageqcpp::ByteStream& bs);
/** @brief Serialize to ByteStream */
EXPORT virtual int serialize(messageqcpp::ByteStream& bs);
DebugDDLStatement(uint32_t debugLevel);
EXPORT DebugDDLStatement();
/** @brief Dump to stdout. */
EXPORT std::ostream& put(std::ostream& os) const;
virtual ~DebugDDLStatement()
{
}
uint32_t fDebugLevel;
};
/** @brief TruncTableStatement represents the drop table operation /** @brief TruncTableStatement represents the drop table operation
*/ */
struct TruncTableStatement : public SqlStatement struct TruncTableStatement : public SqlStatement

40
dbcon/ddlpackage/debugstatement.cpp Normal file
View File

@ -0,0 +1,40 @@
/* Copyright (C) 2024 MariaDB Corporation
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; version 2 of
the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
MA 02110-1301, USA. */
#define DDLPKG_DLLEXPORT
#include "ddlpkg.h"
#undef DDLPKG_DLLEXPORT
using namespace std;
namespace ddlpackage
{
DebugDDLStatement::DebugDDLStatement(uint32_t debugLevel)
: fDebugLevel(debugLevel)
{
}
DebugDDLStatement::DebugDDLStatement()
: fDebugLevel(0)
{
}
ostream& DebugDDLStatement::put(ostream& os) const
{
os << "DDL debug level: " << fDebugLevel << endl;
return os;
}
} // namespace ddlpackage

15
dbcon/ddlpackage/serialize.cpp
View File

@ -429,6 +429,21 @@ int DropTableStatement::serialize(ByteStream& bytestream)
return ret; return ret;
} }
/** @brief Construct from Bytestream */
int DebugDDLStatement::unserialize(ByteStream& bytestream)
{
bytestream >> fDebugLevel;
return 1;
}
/** @brief Serialize to ByteStream */
int DebugDDLStatement::serialize(ByteStream& bytestream)
{
bytestream << (quadbyte)DDL_DEBUG_STATEMENT;
bytestream << fDebugLevel;
return 1;
}
/////////////////////////////////////// ///////////////////////////////////////
/// TruncTableStatement Serialization /// TruncTableStatement Serialization
/////////////////////////////////////// ///////////////////////////////////////

25
dbcon/ddlpackageproc/droppartitionprocessor.cpp
View File

@ -122,6 +122,7 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
auto stmt = formatStatementString(dropPartitionStmt->fSql, dropPartitionStmt->fTableName->fSchema, auto stmt = formatStatementString(dropPartitionStmt->fSql, dropPartitionStmt->fTableName->fSchema,
dropPartitionStmt->fTableName->fName, dropPartitionStmt->fPartitions); dropPartitionStmt->fTableName->fName, dropPartitionStmt->fPartitions);
SQLLogger logger(stmt, fDDLLoggingId, sessionID, txnID.id); SQLLogger logger(stmt, fDDLLoggingId, sessionID, txnID.id);
ostringstream debugLog;
try try
{ {
@ -226,6 +227,8 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
return result; return result;
} }
} }
debugLog << "DropPartitionProcessor: got table lock (id) " << uniqueID << " for table (OID)"
<< roPair.objnum << endl;
// 1. Get the OIDs for the columns // 1. Get the OIDs for the columns
// 2. Get the OIDs for the dictionaries // 2. Get the OIDs for the dictionaries
@ -241,9 +244,18 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
tableColRidList = systemCatalogPtr->columnRIDs(userTableName); tableColRidList = systemCatalogPtr->columnRIDs(userTableName);
tableAuxColOid = systemCatalogPtr->tableAUXColumnOID(userTableName); tableAuxColOid = systemCatalogPtr->tableAUXColumnOID(userTableName);
dictOIDList = systemCatalogPtr->dictOIDs(userTableName); dictOIDList = systemCatalogPtr->dictOIDs(userTableName);
debugLog << "DropPartitionProcessor:" << endl;
debugLog << "column RIDS: ";
for (const auto& rid : tableColRidList)
debugLog << rid.objnum << " ";
debugLog << endl;
debugLog << "dict OIDS: ";
for (const auto& dictOid : dictOIDList)
debugLog << dictOid.dictOID << " ";
debugLog << endl;
// Save qualified tablename, all column, dictionary OIDs, and transaction ID into a file in ASCII format // Save qualified tablename, all column, dictionary OIDs, and transaction ID into a file in ASCII format
for (unsigned i = 0; i < tableColRidList.size(); i++) for (unsigned i = 0; i < tableColRidList.size(); i++)
{ {
@ -271,6 +283,12 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
{ {
throw std::runtime_error(emsg); throw std::runtime_error(emsg);
} }
debugLog << "DropPartitionProcessor: marked partitions for deletion:" << endl;
for (const auto& partition : dropPartitionStmt->fPartitions)
debugLog << "dbroot: " << partition.dbroot << " physical parition: " << partition.pp
<< " segment: " << partition.seg << endl;
VERBOSE_INFO(debugLog.str());
debugLog.clear();
set<BRM::LogicalPartition> markedPartitions; set<BRM::LogicalPartition> markedPartitions;
set<BRM::LogicalPartition> outOfServicePartitions; set<BRM::LogicalPartition> outOfServicePartitions;
@ -298,14 +316,16 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
// Save the oids to a file // Save the oids to a file
createWritePartitionLogFile(roPair.objnum, markedPartitions, oidList, uniqueId); createWritePartitionLogFile(roPair.objnum, markedPartitions, oidList, uniqueId);
VERBOSE_INFO("Removing files"); VERBOSE_INFO("Removing parition files");
removePartitionFiles(oidList, markedPartitions, uniqueId); removePartitionFiles(oidList, markedPartitions, uniqueId);
// Flush PrimProc cache for those lbids // Flush PrimProc cache for those lbids
VERBOSE_INFO("Flushing paritions");
rc = cacheutils::flushPartition(oidList, markedPartitions); rc = cacheutils::flushPartition(oidList, markedPartitions);
// Remove the partition from extent map // Remove the partition from extent map
emsg.clear(); emsg.clear();
rc = fDbrm->deletePartition(oidList, dropPartitionStmt->fPartitions, emsg); rc = fDbrm->deletePartition(oidList, dropPartitionStmt->fPartitions, emsg);
VERBOSE_INFO("DropPartitionProcessor: partitions removed");
if (rc != 0) if (rc != 0)
throw std::runtime_error(emsg); throw std::runtime_error(emsg);
@ -390,6 +410,7 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
} }
fSessionManager.committed(txnID); fSessionManager.committed(txnID);
VERBOSE_INFO("DropPartitionProcessor:: commited");
return result; return result;
} }

273
dbcon/mysql/ha_mcs_client_udfs.cpp
View File

@ -38,6 +38,9 @@ using namespace oam;
#include "errorids.h" #include "errorids.h"
using namespace logging; using namespace logging;
#include "ddlpkg.h"
#include "calpontsystemcatalog.h"
// #include "resourcemanager.h" // #include "resourcemanager.h"
#include "columnstoreversion.h" #include "columnstoreversion.h"
@ -111,8 +114,8 @@ extern "C"
return 0; return 0;
} }
const char* mcssetparms(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcssetparms(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
char parameter[MAXSTRINGLENGTH]; char parameter[MAXSTRINGLENGTH];
char valuestr[MAXSTRINGLENGTH]; char valuestr[MAXSTRINGLENGTH];
@ -180,29 +183,27 @@ extern "C"
return result; return result;
} }
my_bool mcssetparms_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
my_bool mcssetparms_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return setparms_init(initid, args, message, "MCSSETPARMS"); return setparms_init(initid, args, message, "MCSSETPARMS");
} }
void mcssetparms_deinit(UDF_INIT* initid) void mcssetparms_deinit(UDF_INIT* initid)
{ {
} }
const char* calsetparms(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
const char* calsetparms(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, char* is_null, char* error)
char* is_null, char* error)
{ {
return mcssetparms(initid, args, result, length, is_null, error); return mcssetparms(initid, args, result, length, is_null, error);
} }
my_bool calsetparms_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calsetparms_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return setparms_init(initid, args, message, "CALSETPARMS"); return setparms_init(initid, args, message, "CALSETPARMS");
} }
void calsetparms_deinit(UDF_INIT* initid) void calsetparms_deinit(UDF_INIT* initid)
{ {
} }
@ -220,8 +221,8 @@ extern "C"
return 0; return 0;
} }
const char* mcsgetstats(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcsgetstats(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
if (get_fe_conn_info_ptr() == NULL) if (get_fe_conn_info_ptr() == NULL)
{ {
@ -247,28 +248,27 @@ extern "C"
return result; return result;
} }
my_bool mcsgetstats_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
my_bool mcsgetstats_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getstats_init(initid, args, message, "MCSGETSTATS"); return getstats_init(initid, args, message, "MCSGETSTATS");
} }
void mcsgetstats_deinit(UDF_INIT* initid) void mcsgetstats_deinit(UDF_INIT* initid)
{ {
} }
const char* calgetstats(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calgetstats(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcsgetstats(initid, args, result, length, is_null, error); return mcsgetstats(initid, args, result, length, is_null, error);
} }
my_bool calgetstats_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calgetstats_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getstats_init(initid, args, message, "CALGETSTATS"); return getstats_init(initid, args, message, "CALGETSTATS");
} }
void calgetstats_deinit(UDF_INIT* initid) void calgetstats_deinit(UDF_INIT* initid)
{ {
} }
@ -283,7 +283,7 @@ extern "C"
return 0; return 0;
} }
long long mcssettrace(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcssettrace(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
if (get_fe_conn_info_ptr() == NULL) if (get_fe_conn_info_ptr() == NULL)
{ {
@ -301,31 +301,31 @@ extern "C"
return oldTrace; return oldTrace;
} }
my_bool mcssettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcssettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return settrace_init(initid, args, message, "MCSSETTRACE"); return settrace_init(initid, args, message, "MCSSETTRACE");
} }
void mcssettrace_deinit(UDF_INIT* initid) void mcssettrace_deinit(UDF_INIT* initid)
{ {
} }
long long calsettrace(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long calsettrace(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
return mcssettrace(initid, args, is_null, error); return mcssettrace(initid, args, is_null, error);
} }
my_bool calsettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calsettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return settrace_init(initid, args, message, "CALSETTRACE"); return settrace_init(initid, args, message, "CALSETTRACE");
} }
void calsettrace_deinit(UDF_INIT* initid) void calsettrace_deinit(UDF_INIT* initid)
{ {
} }
// Return 1 if system is ready for reads or 0 if not. // Return 1 if system is ready for reads or 0 if not.
long long mcssystemready(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcssystemready(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
long long rtn = 0; long long rtn = 0;
Oam oam; Oam oam;
@ -346,17 +346,17 @@ extern "C"
return rtn; return rtn;
} }
my_bool mcssystemready_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcssystemready_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return 0; return 0;
} }
void mcssystemready_deinit(UDF_INIT* initid) void mcssystemready_deinit(UDF_INIT* initid)
{ {
} }
// Return non-zero if system is read only; 0 if writeable // Return non-zero if system is read only; 0 if writeable
long long mcssystemreadonly(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcssystemreadonly(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
long long rtn = 0; long long rtn = 0;
DBRM dbrm(true); DBRM dbrm(true);
@ -382,17 +382,17 @@ extern "C"
return rtn; return rtn;
} }
my_bool mcssystemreadonly_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcssystemreadonly_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return 0; return 0;
} }
void mcssystemreadonly_deinit(UDF_INIT* initid) void mcssystemreadonly_deinit(UDF_INIT* initid)
{ {
} }
// Return non-zero if this is the primary UM; 0 if not primary // Return non-zero if this is the primary UM; 0 if not primary
long long mcssystemprimary(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcssystemprimary(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
long long rtn = 0; long long rtn = 0;
Oam oam; Oam oam;
@ -424,12 +424,12 @@ extern "C"
return rtn; return rtn;
} }
my_bool mcssystemprimary_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcssystemprimary_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return 0; return 0;
} }
void mcssystemprimary_deinit(UDF_INIT* initid) void mcssystemprimary_deinit(UDF_INIT* initid)
{ {
} }
@ -462,13 +462,13 @@ extern "C"
return 0; return 0;
} }
my_bool mcsviewtablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcsviewtablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return viewtablelock_init(initid, args, message, "MCSVIEWTABLELOCK"); return viewtablelock_init(initid, args, message, "MCSVIEWTABLELOCK");
} }
const char* mcsviewtablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcsviewtablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
THD* thd = current_thd; THD* thd = current_thd;
@ -522,22 +522,22 @@ extern "C"
return result; return result;
} }
void mcsviewtablelock_deinit(UDF_INIT* initid) void mcsviewtablelock_deinit(UDF_INIT* initid)
{ {
} }
my_bool calviewtablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calviewtablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return viewtablelock_init(initid, args, message, "CALVIEWTABLELOCK"); return viewtablelock_init(initid, args, message, "CALVIEWTABLELOCK");
} }
const char* calviewtablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calviewtablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcsviewtablelock(initid, args, result, length, is_null, error); return mcsviewtablelock(initid, args, result, length, is_null, error);
} }
void calviewtablelock_deinit(UDF_INIT* initid) void calviewtablelock_deinit(UDF_INIT* initid)
{ {
} }
@ -555,14 +555,88 @@ extern "C"
return 0; return 0;
} }
my_bool mcs_set_ddldebug_level_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname)
{
if ((args->arg_count != 1) || (args->arg_type[0] != INT_RESULT) ||
*reinterpret_cast<uint32_t*>(args->args[0]) > 3)
{
sprintf(message, "%s() requires one integer `debug level` ", funcname);
return 1;
}
my_bool mcscleartablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message) initid->maybe_null = 0;
initid->max_length = 5;
return 0;
}
const char* mcs_set_ddldebug_level(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length)
{
uint32_t level = *reinterpret_cast<uint32_t*>(args->args[0]);
std::unique_ptr<ddlpackage::DebugDDLStatement> stmt =
std::make_unique<ddlpackage::DebugDDLStatement>(level);
stmt->fSessionID = execplan::CalpontSystemCatalog::idb_tid2sid(current_thd->thread_id);
MessageQueueClient mq("DDLProc");
ByteStream bytestream;
bytestream << (uint32_t)stmt->fSessionID;
stmt->serialize(bytestream);
ByteStream::byte b = 0;
THD* thd = current_thd;
string emsg;
mq.write(bytestream);
try
{
bytestream = mq.read();
if (bytestream.length() == 0)
{
thd->get_stmt_da()->set_overwrite_status(true);
thd->raise_error_printf(ER_INTERNAL_ERROR, "Lost connection to DDLProc");
std::memcpy(result, "Error", 5);
*length = 5;
return result;
}
else
{
bytestream >> b;
bytestream >> emsg;
}
}
catch (runtime_error&)
{
thd->get_stmt_da()->set_overwrite_status(true);
thd->raise_error_printf(ER_INTERNAL_ERROR, "Lost connection to DDLProc");
std::memcpy(result, "Error", 5);
*length = 5;
return result;
}
catch (...)
{
thd->get_stmt_da()->set_overwrite_status(true);
thd->raise_error_printf(ER_INTERNAL_ERROR, "Unknown error caught");
std::memcpy(result, "Error", 5);
*length = 5;
return result;
}
*length = 2;
std::memcpy(result, "Ok", 2);
return result;
}
void mcs_set_ddldebug_level_deinit(UDF_INIT* initid)
{
}
my_bool mcscleartablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return cleartablelock_init(initid, args, message, "MCSCLEARTABLELOCK"); return cleartablelock_init(initid, args, message, "MCSCLEARTABLELOCK");
} }
const char* mcscleartablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcscleartablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
if (get_fe_conn_info_ptr() == NULL) if (get_fe_conn_info_ptr() == NULL)
{ {
@ -588,22 +662,22 @@ extern "C"
return result; return result;
} }
void mcscleartablelock_deinit(UDF_INIT* initid) void mcscleartablelock_deinit(UDF_INIT* initid)
{ {
} }
my_bool calcleartablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calcleartablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return cleartablelock_init(initid, args, message, "CALCLEARTABLELOCK"); return cleartablelock_init(initid, args, message, "CALCLEARTABLELOCK");
} }
const char* calcleartablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calcleartablelock(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcscleartablelock(initid, args, result, length, is_null, error); return mcscleartablelock(initid, args, result, length, is_null, error);
} }
void calcleartablelock_deinit(UDF_INIT* initid) void calcleartablelock_deinit(UDF_INIT* initid)
{ {
} }
@ -636,13 +710,12 @@ extern "C"
return 0; return 0;
} }
my_bool mcslastinsertid_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
my_bool mcslastinsertid_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return lastinsertid_init(initid, args, message, "MCSLASTINSERTID"); return lastinsertid_init(initid, args, message, "MCSLASTINSERTID");
} }
long long mcslastinsertid(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcslastinsertid(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
THD* thd = current_thd; THD* thd = current_thd;
@ -702,21 +775,21 @@ extern "C"
return (nextVal - 1); return (nextVal - 1);
} }
void mcslastinsertid_deinit(UDF_INIT* initid) void mcslastinsertid_deinit(UDF_INIT* initid)
{ {
} }
my_bool callastinsertid_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool callastinsertid_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return lastinsertid_init(initid, args, message, "CALLASTINSERTID"); return lastinsertid_init(initid, args, message, "CALLASTINSERTID");
} }
long long callastinsertid(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long callastinsertid(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
return mcslastinsertid(initid, args, is_null, error); return mcslastinsertid(initid, args, is_null, error);
} }
void callastinsertid_deinit(UDF_INIT* initid) void callastinsertid_deinit(UDF_INIT* initid)
{ {
} }
@ -731,41 +804,41 @@ extern "C"
return 0; return 0;
} }
void mcsflushcache_deinit(UDF_INIT* initid) void mcsflushcache_deinit(UDF_INIT* initid)
{ {
} }
long long mcsflushcache(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcsflushcache(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
return static_cast<long long>(cacheutils::flushPrimProcCache()); return static_cast<long long>(cacheutils::flushPrimProcCache());
} }
my_bool mcsflushcache_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcsflushcache_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return flushcache_init(initid, args, message, "MCSFLUSHCACHE"); return flushcache_init(initid, args, message, "MCSFLUSHCACHE");
} }
void calflushcache_deinit(UDF_INIT* initid) void calflushcache_deinit(UDF_INIT* initid)
{ {
} }
long long calflushcache(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long calflushcache(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
return mcsflushcache(initid, args, is_null, error); return mcsflushcache(initid, args, is_null, error);
} }
my_bool calflushcache_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calflushcache_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return flushcache_init(initid, args, message, "CALFLUSHCACHE"); return flushcache_init(initid, args, message, "CALFLUSHCACHE");
} }
static const unsigned long TraceSize = 16 * 1024; static const unsigned long TraceSize = 16 * 1024;
// mysqld will call this with only 766 bytes available in result no matter what we asked for in // mysqld will call this with only 766 bytes available in result no matter what we asked for in
// calgettrace_init() // calgettrace_init()
// if we return a pointer that is not result, mysqld will take our pointer and use it, freeing up result // if we return a pointer that is not result, mysqld will take our pointer and use it, freeing up result
const char* mcsgettrace(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcsgettrace(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
const std::string* msgp; const std::string* msgp;
int flags = 0; int flags = 0;
@ -808,7 +881,7 @@ extern "C"
return msgp->c_str(); return msgp->c_str();
} }
my_bool gettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname) my_bool gettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname)
{ {
#if 0 #if 0
@ -825,27 +898,27 @@ extern "C"
return 0; return 0;
} }
my_bool mcsgettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcsgettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return gettrace_init(initid, args, message, "MCSGETTRACE"); return gettrace_init(initid, args, message, "MCSGETTRACE");
} }
void mcsgettrace_deinit(UDF_INIT* initid) void mcsgettrace_deinit(UDF_INIT* initid)
{ {
} }
const char* calgettrace(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calgettrace(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcsgettrace(initid, args, result, length, is_null, error); return mcsgettrace(initid, args, result, length, is_null, error);
} }
my_bool calgettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calgettrace_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return gettrace_init(initid, args, message, "CALGETTRACE"); return gettrace_init(initid, args, message, "CALGETTRACE");
} }
void calgettrace_deinit(UDF_INIT* initid) void calgettrace_deinit(UDF_INIT* initid)
{ {
} }
@ -860,8 +933,8 @@ extern "C"
return 0; return 0;
} }
const char* mcsgetversion(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcsgetversion(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
std::string version(columnstore_version); std::string version(columnstore_version);
*length = version.size(); *length = version.size();
@ -869,27 +942,27 @@ extern "C"
return result; return result;
} }
my_bool mcsgetversion_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcsgetversion_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getversion_init(initid, args, message, "MCSGETVERSION"); return getversion_init(initid, args, message, "MCSGETVERSION");
} }
void mcsgetversion_deinit(UDF_INIT* initid) void mcsgetversion_deinit(UDF_INIT* initid)
{ {
} }
const char* calgetversion(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calgetversion(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcsgetversion(initid, args, result, length, is_null, error); return mcsgetversion(initid, args, result, length, is_null, error);
} }
my_bool calgetversion_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calgetversion_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getversion_init(initid, args, message, "CALGETVERSION"); return getversion_init(initid, args, message, "CALGETVERSION");
} }
void calgetversion_deinit(UDF_INIT* initid) void calgetversion_deinit(UDF_INIT* initid)
{ {
} }
@ -904,8 +977,8 @@ extern "C"
return 0; return 0;
} }
const char* mcsgetsqlcount(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* mcsgetsqlcount(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
if (get_fe_conn_info_ptr() == NULL) if (get_fe_conn_info_ptr() == NULL)
{ {
@ -946,37 +1019,37 @@ extern "C"
return result; return result;
} }
my_bool mcsgetsqlcount_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcsgetsqlcount_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getstats_init(initid, args, message, "MCSGETSQLCOUNT"); return getstats_init(initid, args, message, "MCSGETSQLCOUNT");
} }
void mcsgetsqlcount_deinit(UDF_INIT* initid) void mcsgetsqlcount_deinit(UDF_INIT* initid)
{ {
} }
const char* calgetsqlcount(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calgetsqlcount(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
return mcsgetsqlcount(initid, args, result, length, is_null, error); return mcsgetsqlcount(initid, args, result, length, is_null, error);
} }
my_bool calgetsqlcount_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calgetsqlcount_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
return getstats_init(initid, args, message, "CALGETSQLCOUNT"); return getstats_init(initid, args, message, "CALGETSQLCOUNT");
} }
void calgetsqlcount_deinit(UDF_INIT* initid) void calgetsqlcount_deinit(UDF_INIT* initid)
{ {
} }
long long mcs_emindex_size(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcs_emindex_size(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
DBRM dbrm; DBRM dbrm;
return dbrm.EMIndexShmemSize(); return dbrm.EMIndexShmemSize();
} }
my_bool mcs_emindex_size_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcs_emindex_size_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
if (args->arg_count != 0) if (args->arg_count != 0)
{ {
@ -987,17 +1060,17 @@ extern "C"
return 0; return 0;
} }
void mcs_emindex_size_deinit(UDF_INIT* initid) void mcs_emindex_size_deinit(UDF_INIT* initid)
{ {
} }
long long mcs_emindex_free(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error) long long mcs_emindex_free(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
{ {
DBRM dbrm; DBRM dbrm;
return dbrm.EMIndexShmemFree(); return dbrm.EMIndexShmemFree();
} }
my_bool mcs_emindex_free_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool mcs_emindex_free_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
if (args->arg_count != 0) if (args->arg_count != 0)
{ {
@ -1008,7 +1081,7 @@ extern "C"
return 0; return 0;
} }
void mcs_emindex_free_deinit(UDF_INIT* initid) void mcs_emindex_free_deinit(UDF_INIT* initid)
{ {
} }

64
dbcon/mysql/ha_mcs_partition.cpp
View File

@ -546,7 +546,7 @@ extern "C"
* CalShowPartitions * CalShowPartitions
*/ */
my_bool calshowpartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calshowpartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
if (args->arg_count < 2 || args->arg_count > 3 || args->arg_type[0] != STRING_RESULT || if (args->arg_count < 2 || args->arg_count > 3 || args->arg_type[0] != STRING_RESULT ||
args->arg_type[1] != STRING_RESULT || (args->arg_count == 3 && args->arg_type[2] != STRING_RESULT)) args->arg_type[1] != STRING_RESULT || (args->arg_count == 3 && args->arg_type[2] != STRING_RESULT))
@ -567,13 +567,13 @@ extern "C"
return 0; return 0;
} }
void calshowpartitions_deinit(UDF_INIT* initid) void calshowpartitions_deinit(UDF_INIT* initid)
{ {
delete[] initid->ptr; delete[] initid->ptr;
} }
const char* calshowpartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calshowpartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
BRM::DBRM::refreshShm(); BRM::DBRM::refreshShm();
DBRM em; DBRM em;
@ -701,7 +701,7 @@ extern "C"
* CalDisablePartitions * CalDisablePartitions
*/ */
my_bool caldisablepartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool caldisablepartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -735,8 +735,8 @@ extern "C"
return 0; return 0;
} }
const char* caldisablepartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* caldisablepartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
CalpontSystemCatalog::TableName tableName; CalpontSystemCatalog::TableName tableName;
set<LogicalPartition> partitionNums; set<LogicalPartition> partitionNums;
@ -772,7 +772,7 @@ extern "C"
return result; return result;
} }
void caldisablepartitions_deinit(UDF_INIT* initid) void caldisablepartitions_deinit(UDF_INIT* initid)
{ {
} }
@ -780,7 +780,7 @@ extern "C"
* CalEnablePartitions * CalEnablePartitions
*/ */
my_bool calenablepartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calenablepartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -814,8 +814,8 @@ extern "C"
return 0; return 0;
} }
const char* calenablepartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* calenablepartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
CalpontSystemCatalog::TableName tableName; CalpontSystemCatalog::TableName tableName;
string errMsg; string errMsg;
@ -851,7 +851,7 @@ extern "C"
return result; return result;
} }
void calenablepartitions_deinit(UDF_INIT* initid) void calenablepartitions_deinit(UDF_INIT* initid)
{ {
} }
@ -859,7 +859,7 @@ extern "C"
* CalDropPartitions * CalDropPartitions
*/ */
my_bool caldroppartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool caldroppartitions_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -893,8 +893,8 @@ extern "C"
return 0; return 0;
} }
const char* caldroppartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length, const char* caldroppartitions(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
char* is_null, char* error) char* is_null, char* error)
{ {
CalpontSystemCatalog::TableName tableName; CalpontSystemCatalog::TableName tableName;
string errMsg; string errMsg;
@ -930,7 +930,7 @@ extern "C"
return result; return result;
} }
void caldroppartitions_deinit(UDF_INIT* initid) void caldroppartitions_deinit(UDF_INIT* initid)
{ {
} }
@ -938,11 +938,11 @@ extern "C"
* CalDropPartitionsByValue * CalDropPartitionsByValue
*/ */
void caldroppartitionsbyvalue_deinit(UDF_INIT* initid) void caldroppartitionsbyvalue_deinit(UDF_INIT* initid)
{ {
} }
my_bool caldroppartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool caldroppartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -979,8 +979,8 @@ extern "C"
return 0; return 0;
} }
const char* caldroppartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, const char* caldroppartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
unsigned long* length, char* is_null, char* error) char* is_null, char* error)
{ {
string msg; string msg;
CalpontSystemCatalog::TableName tableName; CalpontSystemCatalog::TableName tableName;
@ -1005,11 +1005,11 @@ extern "C"
* CalDisablePartitionsByValue * CalDisablePartitionsByValue
*/ */
void caldisablepartitionsbyvalue_deinit(UDF_INIT* initid) void caldisablepartitionsbyvalue_deinit(UDF_INIT* initid)
{ {
} }
my_bool caldisablepartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool caldisablepartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -1042,8 +1042,8 @@ extern "C"
return 0; return 0;
} }
const char* caldisablepartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, const char* caldisablepartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result,
unsigned long* length, char* is_null, char* error) unsigned long* length, char* is_null, char* error)
{ {
string msg; string msg;
set<LogicalPartition> partSet; set<LogicalPartition> partSet;
@ -1067,11 +1067,11 @@ extern "C"
/** /**
* CalEnablePartitionsByValue * CalEnablePartitionsByValue
*/ */
void calenablepartitionsbyvalue_deinit(UDF_INIT* initid) void calenablepartitionsbyvalue_deinit(UDF_INIT* initid)
{ {
} }
my_bool calenablepartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calenablepartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -1104,8 +1104,8 @@ extern "C"
return 0; return 0;
} }
const char* calenablepartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, const char* calenablepartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result,
unsigned long* length, char* is_null, char* error) unsigned long* length, char* is_null, char* error)
{ {
string msg; string msg;
set<LogicalPartition> partSet; set<LogicalPartition> partSet;
@ -1129,7 +1129,7 @@ extern "C"
/** /**
* CalShowPartitionsByValue * CalShowPartitionsByValue
*/ */
my_bool calshowpartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message) my_bool calshowpartitionsbyvalue_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
{ {
bool err = false; bool err = false;
@ -1162,13 +1162,13 @@ extern "C"
return 0; return 0;
} }
void calshowpartitionsbyvalue_deinit(UDF_INIT* initid) void calshowpartitionsbyvalue_deinit(UDF_INIT* initid)
{ {
delete[] initid->ptr; delete[] initid->ptr;
} }
const char* calshowpartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, const char* calshowpartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
unsigned long* length, char* is_null, char* error) char* is_null, char* error)
{ {
BRM::DBRM::refreshShm(); BRM::DBRM::refreshShm();
DBRM em; DBRM em;

1
dbcon/mysql/install_mcs_mysql.sh.in
View File

@ -79,6 +79,7 @@ CREATE OR REPLACE FUNCTION caldisablepartitionsbyvalue RETURNS STRING SONAME 'ha
CREATE OR REPLACE FUNCTION calenablepartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so'; CREATE OR REPLACE FUNCTION calenablepartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so';
CREATE OR REPLACE FUNCTION calshowpartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so'; CREATE OR REPLACE FUNCTION calshowpartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so';
CREATE OR REPLACE AGGREGATE FUNCTION moda RETURNS STRING SONAME 'libregr_mysql.so'; CREATE OR REPLACE AGGREGATE FUNCTION moda RETURNS STRING SONAME 'libregr_mysql.so';
CREATE OR REPLACE FUNCTION mcs_set_ddldebug_level RETURNS STRING SONAME 'ha_columnstore.so';
CREATE DATABASE IF NOT EXISTS infinidb_querystats; CREATE DATABASE IF NOT EXISTS infinidb_querystats;
CREATE TABLE IF NOT EXISTS infinidb_querystats.querystats CREATE TABLE IF NOT EXISTS infinidb_querystats.querystats

34
ddlproc/ddlprocessor.cpp
View File

@ -75,8 +75,9 @@ void cleanPMSysCache()
class PackageHandler class PackageHandler
{ {
public: public:
PackageHandler(QueryTeleClient qtc, DBRM* dbrm, messageqcpp::IOSocket& ios, bool concurrentSupport) PackageHandler(QueryTeleClient qtc, DBRM* dbrm, messageqcpp::IOSocket& ios, bool concurrentSupport,
: fIos(ios), fDbrm(dbrm), fQtc(qtc), fConcurrentSupport(concurrentSupport) uint32_t* debugLevel)
: fIos(ios), fDbrm(dbrm), fQtc(qtc), fConcurrentSupport(concurrentSupport), fDebugLevel(debugLevel)
{ {
} }
@ -490,6 +491,8 @@ class PackageHandler
boost::scoped_ptr<CreateTableProcessor> processor(new CreateTableProcessor(fDbrm)); boost::scoped_ptr<CreateTableProcessor> processor(new CreateTableProcessor(fDbrm));
processor->fTxnid.id = fTxnid.id; processor->fTxnid.id = fTxnid.id;
processor->fTxnid.valid = true; processor->fTxnid.valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
// cout << "create table using txnid " << fTxnid.id << endl; // cout << "create table using txnid " << fTxnid.id << endl;
QueryTeleStats qts; QueryTeleStats qts;
@ -524,6 +527,8 @@ class PackageHandler
boost::scoped_ptr<AlterTableProcessor> processor(new AlterTableProcessor(fDbrm)); boost::scoped_ptr<AlterTableProcessor> processor(new AlterTableProcessor(fDbrm));
processor->fTxnid.id = fTxnid.id; processor->fTxnid.id = fTxnid.id;
processor->fTxnid.valid = true; processor->fTxnid.valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
QueryTeleStats qts; QueryTeleStats qts;
qts.query_uuid = QueryTeleClient::genUUID(); qts.query_uuid = QueryTeleClient::genUUID();
@ -560,6 +565,8 @@ class PackageHandler
processor->fTxnid.id = fTxnid.id; processor->fTxnid.id = fTxnid.id;
processor->fTxnid.valid = true; processor->fTxnid.valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
QueryTeleStats qts; QueryTeleStats qts;
qts.query_uuid = QueryTeleClient::genUUID(); qts.query_uuid = QueryTeleClient::genUUID();
@ -595,6 +602,8 @@ class PackageHandler
processor->fTxnid.id = fTxnid.id; processor->fTxnid.id = fTxnid.id;
processor->fTxnid.valid = true; processor->fTxnid.valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
QueryTeleStats qts; QueryTeleStats qts;
qts.query_uuid = QueryTeleClient::genUUID(); qts.query_uuid = QueryTeleClient::genUUID();
@ -628,6 +637,9 @@ class PackageHandler
boost::scoped_ptr<MarkPartitionProcessor> processor(new MarkPartitionProcessor(fDbrm)); boost::scoped_ptr<MarkPartitionProcessor> processor(new MarkPartitionProcessor(fDbrm));
(processor->fTxnid).id = fTxnid.id; (processor->fTxnid).id = fTxnid.id;
(processor->fTxnid).valid = true; (processor->fTxnid).valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
result = processor->processPackage(&markPartitionStmt); result = processor->processPackage(&markPartitionStmt);
systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID); systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID);
systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID | 0x80000000); systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID | 0x80000000);
@ -643,6 +655,9 @@ class PackageHandler
boost::scoped_ptr<RestorePartitionProcessor> processor(new RestorePartitionProcessor(fDbrm)); boost::scoped_ptr<RestorePartitionProcessor> processor(new RestorePartitionProcessor(fDbrm));
(processor->fTxnid).id = fTxnid.id; (processor->fTxnid).id = fTxnid.id;
(processor->fTxnid).valid = true; (processor->fTxnid).valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
result = processor->processPackage(&restorePartitionStmt); result = processor->processPackage(&restorePartitionStmt);
systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID); systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID);
systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID | 0x80000000); systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID | 0x80000000);
@ -658,12 +673,24 @@ class PackageHandler
boost::scoped_ptr<DropPartitionProcessor> processor(new DropPartitionProcessor(fDbrm)); boost::scoped_ptr<DropPartitionProcessor> processor(new DropPartitionProcessor(fDbrm));
(processor->fTxnid).id = fTxnid.id; (processor->fTxnid).id = fTxnid.id;
(processor->fTxnid).valid = true; (processor->fTxnid).valid = true;
if (fDebugLevel)
processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
result = processor->processPackage(&dropPartitionStmt); result = processor->processPackage(&dropPartitionStmt);
systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID); systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID);
systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID | 0x80000000); systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID | 0x80000000);
} }
break; break;
case ddlpackage::DDL_DEBUG_STATEMENT:
{
DebugDDLStatement stmt;
stmt.unserialize(fByteStream);
if (fDebugLevel)
*fDebugLevel = stmt.fDebugLevel;
}
break;
default: throw UNRECOGNIZED_PACKAGE_TYPE; break; default: throw UNRECOGNIZED_PACKAGE_TYPE; break;
} }
@ -728,6 +755,7 @@ class PackageHandler
QueryTeleClient fQtc; QueryTeleClient fQtc;
oam::OamCache* fOamCache = nullptr; oam::OamCache* fOamCache = nullptr;
bool fConcurrentSupport; bool fConcurrentSupport;
uint32_t* fDebugLevel{nullptr};
}; };
} // namespace } // namespace
@ -782,7 +810,7 @@ void DDLProcessor::process()
try try
{ {
ios = fMqServer.accept(); ios = fMqServer.accept();
fDdlPackagepool.invoke(PackageHandler(fQtc, &dbrm, ios, concurrentSupport)); fDdlPackagepool.invoke(PackageHandler(fQtc, &dbrm, ios, concurrentSupport, &debugLevel));
} }
catch (...) catch (...)
{ {

1
ddlproc/ddlprocessor.h
View File

@ -93,6 +93,7 @@ class DDLProcessor
WriteEngine::WEClients* fWEClient; WriteEngine::WEClients* fWEClient;
uint32_t fPMCount; uint32_t fPMCount;
querytele::QueryTeleClient fQtc; querytele::QueryTeleClient fQtc;
uint32_t debugLevel{0};
}; };
} // namespace ddlprocessor } // namespace ddlprocessor