database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-11-03 17:13:17 +03:00
Code Activity
Files
5851e028a83ea0052959b016bb8617aea15bf17f
mariadb-columnstore-engine/writeengine/server/we_ddlcommandproc.cpp
Leonid Fedorov 4a8f5830fd chore(build) update boost to 1.88
2025-05-20 21:43:43 +04:00

4599 lines
128 KiB
C++
Raw Blame History

/* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016 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. */
// $Id: we_ddlcommandproc.cpp 3082 2011-09-26 22:00:38Z chao $
#include <unistd.h>
#include "boost/filesystem.hpp"
#include "boost/scoped_ptr.hpp"
using namespace std;
#include "bytestream.h"
using namespace messageqcpp;
#include "we_messages.h"
#include "we_message_handlers.h"
#include "we_ddlcommon.h"
#include "we_ddlcommandproc.h"
#include "ddlpkg.h"
using namespace ddlpackage;
#include <ctime>
#include "dataconvert.h"
using namespace dataconvert;
// #include "we_brm.h"
namespace fs = boost::filesystem;
#include "cacheutils.h"
#include "IDBDataFile.h"
#include "IDBPolicy.h"
using namespace idbdatafile;
using namespace execplan;
namespace WriteEngine
{
WE_DDLCommandProc::WE_DDLCommandProc()
{
filesPerColumnPartition = 8;
extentsPerSegmentFile = 1;
dbrootCnt = 1;
extentRows = 0x800000;
config::Config* cf = config::Config::makeConfig();
string fpc = cf->getConfig("ExtentMap", "FilesPerColumnPartition");
if (fpc.length() != 0)
filesPerColumnPartition = cf->uFromText(fpc);
// MCOL-4685: remove the option to set more than 2 extents per file (ExtentsPreSegmentFile).
extentsPerSegmentFile = DEFAULT_EXTENTS_PER_SEGMENT_FILE;
string dbct = cf->getConfig("SystemConfig", "DBRootCount");
if (dbct.length() != 0)
dbrootCnt = cf->uFromText(dbct);
}
WE_DDLCommandProc::~WE_DDLCommandProc()
{
}
uint8_t WE_DDLCommandProc::updateSyscolumnNextval(ByteStream& bs, std::string& err)
{
uint32_t columnOid, sessionID;
uint64_t nextVal;
int rc = 0;
bs >> columnOid;
bs >> nextVal;
bs >> sessionID;
uint16_t dbRoot;
BRM::OID_t oid = 1021;
fDbrm.getSysCatDBRoot(oid, dbRoot);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
oids[columnOid] = columnOid;
// oidsToFlush.push_back(columnOid);
if (idbdatafile::IDBPolicy::useHdfs())
fWEWrapper.startTransaction(sessionID);
rc = fWEWrapper.updateNextValue(sessionID, columnOid, nextVal, sessionID, dbRoot);
if (rc != 0)
{
err = "Error in WE::updateNextValue";
rc = 1;
}
if (idbdatafile::IDBPolicy::useHdfs())
{
fWEWrapper.flushDataFiles(rc, sessionID, oids);
fWEWrapper.confirmTransaction(sessionID);
if (rc == 0)
fWEWrapper.endTransaction(sessionID, true);
else
fWEWrapper.endTransaction(sessionID, false);
}
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::writeSystable(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
int txnID, tableOID, tableAUXColumnOID;
uint32_t tableWithAutoi;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> tmp32;
tableOID = tmp32;
bs >> tmp32;
tableAUXColumnOID = tmp32;
bs >> tmp32;
tableWithAutoi = tmp32;
bs >> tmp32;
uint16_t dbroot = tmp32;
ddlpackage::TableDef tableDef;
tableDef.unserialize(bs);
WriteEngine::ColTuple colTuple;
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
WriteEngine::ColTupleList colTuples;
WriteEngine::dictStr dctColTuples;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::ColValueList colValuesList;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DictStrList dctnryValueList;
WriteEngine::RIDList ridList;
CalpontSystemCatalog::TableName tableName;
CalpontSystemCatalog::ROPair sysTableROPair;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
ColumnList columns;
ColumnList::const_iterator column_iterator;
DDLColumn column;
int error = 0;
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSTABLE_TABLE;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
try
{
sysTableROPair = systemCatalogPtr->tableRID(tableName);
getColumnsForTable(sessionID, tableName.schema, tableName.table, columns);
column_iterator = columns.begin();
NullString tmpStr;
while (column_iterator != columns.end())
{
column = *column_iterator;
boost::to_lower(column.tableColName.column);
tmpStr.dropString();
if (TABLENAME_COL == column.tableColName.column)
{
std::string tablename = tableDef.fQualifiedName->fName;
colTuple.data = tablename;
tmpStr.assign(tablename);
}
else if (SCHEMA_COL == column.tableColName.column)
{
std::string schema = tableDef.fQualifiedName->fSchema;
colTuple.data = schema;
tmpStr.assign(schema);
}
else if (OBJECTID_COL == column.tableColName.column)
{
colTuple.data = tableOID;
}
else if (AUXCOLUMNOID_COL == column.tableColName.column)
{
colTuple.data = tableAUXColumnOID;
}
else if (CREATEDATE_COL == column.tableColName.column)
{
time_t t;
struct tm tmp;
Date aDay;
t = time(NULL);
gmtime_r(&t, &tmp);
aDay.year = tmp.tm_year + 1900;
aDay.month = tmp.tm_mon + 1;
aDay.day = tmp.tm_mday;
colTuple.data = *(reinterpret_cast<int*>(&aDay));
}
else if (INIT_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (NEXT_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (AUTOINC_COL == column.tableColName.column)
{
colTuple.data = tableWithAutoi;
}
else
{
colTuple.data = column.colType.getNullValueForType();
}
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.tokenFlag = column.colType.colWidth > 8 ? true : false;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbroot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
dctnryStruct.fColDbRoot = dbroot;
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = column.oid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = column.oid;
}
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
colTuples.push_back(colTuple);
dctColTuples.push_back(tmpStr);
colValuesList.push_back(colTuples);
dctnryStructList.push_back(dctnryStruct);
dctnryValueList.push_back(dctColTuples);
colTuples.pop_back();
dctColTuples.pop_back();
++column_iterator;
}
// fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
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, cscColTypeList, colStructs, colValuesList,
dctnryStructList, dctnryValueList, SYSCOLUMN_BASE);
if (error != WriteEngine::NO_ERROR)
{
if (error == ERR_BRM_WR_VB_ENTRY)
{
throw std::runtime_error("WE: Error writing to BRM.");
}
else
{
WErrorCodes ec;
throw std::runtime_error("WE: Error updating calpontsys.systable:" + ec.errorString(error));
}
}
if (idbdatafile::IDBPolicy::useHdfs())
{
int rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
}
}
catch (exception& ex)
{
err += ex.what();
rc = 1;
}
catch (...)
{
err += "Unknown exception caught";
rc = 1;
}
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
return rc;
}
uint8_t WE_DDLCommandProc::writeCreateSyscolumn(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32, columnSize, dictSize, i;
uint8_t tmp8;
int txnID, colpos;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> columnSize;
// deserialize column Oid and dictionary oid
vector<uint32_t> coloids;
vector<uint32_t> dictoids;
for (i = 0; i < columnSize; ++i)
{
bs >> tmp32;
coloids.push_back(tmp32);
}
bs >> dictSize;
for (i = 0; i < dictSize; ++i)
{
bs >> tmp32;
dictoids.push_back(tmp32);
}
bool alterFlag = 0;
bs >> tmp8;
alterFlag = (tmp8 != 0);
bs >> tmp32;
colpos = tmp32;
bs >> tmp32;
uint16_t dbroot = tmp32;
ddlpackage::TableDef tableDef;
tableDef.unserialize(bs);
WriteEngine::ColTuple colTuple;
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
WriteEngine::ColTupleList colTuples;
WriteEngine::dictStr dctColTuples;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::ColValueList colValuesList;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DictStrList dctnryValueList;
WriteEngine::RIDList ridList;
CalpontSystemCatalog::TableName tableName;
CalpontSystemCatalog::ROPair sysTableROPair;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
ColumnList columns;
ColumnList::const_iterator column_iterator;
DDLColumn column;
int error = 0;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
std::map<uint32_t, uint32_t> oids;
int rc1 = 0;
ColumnDef* colDefPtr = 0;
ColumnDefList::const_iterator iter;
int startPos = colpos;
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSCOLUMN_TABLE;
getColumnsForTable(sessionID, tableName.schema, tableName.table, columns);
unsigned int numCols = columns.size();
// WriteEngine::ColTupleList colList[numCols];
// ColTupleList is NOT POD, so let's try this:
std::vector<WriteEngine::ColTupleList> colList;
// WriteEngine::dictStr dctColList[numCols];
std::vector<WriteEngine::dictStr> dctColList;
ColumnDefList tableDefCols = tableDef.fColumns;
ddlpackage::QualifiedName qualifiedName = *(tableDef.fQualifiedName);
iter = tableDefCols.begin();
// colpos = 0;
NullString tmpStr;
for (unsigned int ii = 0; ii < numCols; ii++)
{
colList.push_back(WriteEngine::ColTupleList());
dctColList.push_back(WriteEngine::dictStr());
}
try
{
unsigned int col = 0;
unsigned int dictcol = 0;
while (iter != tableDefCols.end())
{
colDefPtr = *iter;
DictOID dictOID = {0, 0, 0, 0, 0};
int dataType;
dataType = convertDataType(colDefPtr->fType->fType);
if (dataType == CalpontSystemCatalog::DECIMAL || dataType == CalpontSystemCatalog::UDECIMAL)
{
if (colDefPtr->fType->fPrecision > 38) // precision cannot be over 38.
{
ostringstream os;
os << "Syntax error: The maximum precision (total number of digits) that can be specified is 38";
throw std::runtime_error(os.str());
}
else if (colDefPtr->fType->fPrecision < colDefPtr->fType->fScale)
{
ostringstream os;
os << "Syntax error: scale should be less than precision, precision: "
<< colDefPtr->fType->fPrecision << " scale: " << colDefPtr->fType->fScale;
throw std::runtime_error(os.str());
}
colDefPtr->convertDecimal();
}
bool hasDict = false;
if ((dataType == CalpontSystemCatalog::CHAR && colDefPtr->fType->fLength > 8) ||
(dataType == CalpontSystemCatalog::VARCHAR && colDefPtr->fType->fLength > 7) ||
(dataType == CalpontSystemCatalog::VARBINARY && colDefPtr->fType->fLength > 7) ||
(dataType == CalpontSystemCatalog::BLOB && colDefPtr->fType->fLength > 7) ||
(dataType == CalpontSystemCatalog::TEXT && colDefPtr->fType->fLength > 7))
{
hasDict = true;
dictOID.compressionType = colDefPtr->fType->fCompressiontype;
dictOID.colWidth = colDefPtr->fType->fLength;
dictOID.dictOID = dictoids[dictcol];
dictcol++;
//@Bug 2534. Take away the limit of 255 and set the limit to 8000.
if ((colDefPtr->fType->fLength > 8000) && (dataType != CalpontSystemCatalog::BLOB) &&
(dataType != CalpontSystemCatalog::TEXT))
{
ostringstream os;
os << "char, varchar and varbinary length may not exceed 8000 bytes";
throw std::runtime_error(os.str());
}
}
else if ((dataType == CalpontSystemCatalog::VARBINARY || dataType == CalpontSystemCatalog::BLOB ||
dataType == CalpontSystemCatalog::TEXT) &&
colDefPtr->fType->fLength <= 7)
{
ostringstream os;
os << "varbinary and blob length may not be less than 8";
throw std::runtime_error(os.str());
}
unsigned int i = 0;
column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
boost::to_lower(column.tableColName.column);
if (SCHEMA_COL == column.tableColName.column)
{
colTuple.data = qualifiedName.fSchema;
tmpStr.assign(qualifiedName.fSchema);
}
else if (TABLENAME_COL == column.tableColName.column)
{
colTuple.data = qualifiedName.fName;
tmpStr.assign(qualifiedName.fName);
}
else if (COLNAME_COL == column.tableColName.column)
{
boost::to_lower(colDefPtr->fName);
colTuple.data = colDefPtr->fName;
tmpStr.assign(colDefPtr->fName);
}
else if (OBJECTID_COL == column.tableColName.column)
{
if (alterFlag)
colTuple.data = coloids[col];
else
colTuple.data = coloids[col];
}
else if (DATATYPE_COL == column.tableColName.column)
{
colTuple.data = dataType;
}
else if (COLUMNLEN_COL == column.tableColName.column)
{
//@Bug 2089 Disallow zero length char and varch column to be created
if (dataType == CalpontSystemCatalog::CHAR || dataType == CalpontSystemCatalog::VARCHAR ||
dataType == CalpontSystemCatalog::VARBINARY || dataType == CalpontSystemCatalog::BLOB ||
dataType == CalpontSystemCatalog::TEXT)
{
if (colDefPtr->fType->fLength <= 0)
{
ostringstream os;
os << "char, varchar and varbinary length must be greater than zero";
throw std::runtime_error(os.str());
}
}
colTuple.data = colDefPtr->fType->fLength;
}
else if (COLUMNPOS_COL == column.tableColName.column)
{
colTuple.data = colpos;
}
else if (DEFAULTVAL_COL == column.tableColName.column)
{
if (colDefPtr->fDefaultValue && !colDefPtr->fDefaultValue->fNull)
{
tmpStr.assign(colDefPtr->fDefaultValue->fValue);
}
else
{
tmpStr.dropString();
}
colTuple.data = tmpStr;
}
else if (NULLABLE_COL == column.tableColName.column)
{
int nullable = 1;
ColumnConstraintList& colConstraints = colDefPtr->fConstraints;
ColumnConstraintList::const_iterator constraint_iter = colConstraints.begin();
while (constraint_iter != colConstraints.end())
{
ColumnConstraintDef* consDefPtr = *constraint_iter;
if (consDefPtr->fConstraintType == ddlpackage::DDL_NOT_NULL)
{
nullable = 0;
break;
}
++constraint_iter;
}
colTuple.data = nullable;
}
else if (SCALE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fScale;
}
else if (PRECISION_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fPrecision;
}
else if (DICTOID_COL == column.tableColName.column)
{
if (hasDict)
{
colTuple.data = dictOID.dictOID;
}
else
{
colTuple.data = column.colType.getNullValueForType();
}
}
else if (LISTOBJID_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (TREEOBJID_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (MINVAL_COL == column.tableColName.column)
{
tmpStr.dropString();
}
else if (MAXVAL_COL == column.tableColName.column)
{
tmpStr.dropString();
}
else if (COMPRESSIONTYPE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fCompressiontype;
}
else if (AUTOINC_COL == column.tableColName.column)
{
// cout << "autoincrement= " << colDefPtr->fType->fAutoincrement << endl;
colTuple.data = colDefPtr->fType->fAutoincrement;
}
else if (NEXTVALUE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fNextvalue;
}
else if (CHARSETNUM_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fCharsetNum;
}
else
{
colTuple.data = column.colType.getNullValueForType();
}
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.tokenFlag = column.colType.colWidth > 8 ? true : false;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbroot;
dctnryStruct.fColDbRoot = dbroot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag) // TODO: XXX: this is copied aplenty. NEED TO REFACTOR.
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = column.oid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = column.oid;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
if (colpos == startPos)
{
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
cscColTypeList.push_back(column.colType);
}
colList[i].push_back(colTuple);
// colList.push_back(WriteEngine::ColTupleList());
// colList.back().push_back(colTuple);
dctColList[i].push_back(tmpStr);
// dctColList.push_back(WriteEngine::dictStr());
// dctColList.back().push_back(tmpStr);
++i;
++column_iterator;
}
++colpos;
col++;
++iter;
}
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
if (0 != colStructs.size())
{
for (unsigned int n = 0; n < numCols; n++)
{
colValuesList.push_back(colList[n]);
dctnryValueList.push_back(dctColList[n]);
}
// fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
error = fWEWrapper.insertColumnRec_SYS(txnID, cscColTypeList, colStructs, colValuesList,
dctnryStructList, dctnryValueList, SYSCOLUMN_BASE);
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error != WriteEngine::NO_ERROR)
{
if (error == ERR_BRM_WR_VB_ENTRY)
{
throw std::runtime_error("writeSysColumnMetaData WE: Error writing to BRM.");
}
else
{
WErrorCodes ec;
throw std::runtime_error("WE: Error updating calpontsys.syscolumn. " + ec.errorString(error));
}
}
else
error = rc1;
}
}
catch (exception& ex)
{
err += ex.what();
rc = 1;
}
catch (...)
{
err += "Unknown exception caught";
rc = 1;
}
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
return rc;
}
uint8_t WE_DDLCommandProc::writeSyscolumn(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32, coloid, dictoid;
int txnID, startPos;
string schema, tablename;
uint8_t tmp8;
bool isAlter = false;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
bs >> coloid;
bs >> dictoid;
bs >> tmp8; // alterFlag
bs >> tmp32;
startPos = tmp32;
isAlter = (tmp8 != 0);
boost::scoped_ptr<ddlpackage::ColumnDef> colDefPtr(new ddlpackage::ColumnDef());
colDefPtr->unserialize(bs);
WriteEngine::ColStruct colStruct;
WriteEngine::ColTuple colTuple;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
WriteEngine::ColTupleList colTuples;
WriteEngine::DctColTupleList dctColTuples;
WriteEngine::ColValueList colValuesList;
WriteEngine::RIDList ridList;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::dictStr dctnryTuple;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DictStrList dctnryValueList;
CalpontSystemCatalog::TableName tableName;
ColumnList columns;
ColumnList::const_iterator column_iterator;
DDLColumn column;
int error = 0;
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSCOLUMN_TABLE;
getColumnsForTable(sessionID, tableName.schema, tableName.table, columns);
unsigned int numCols = columns.size();
// WriteEngine::ColTupleList colList[numCols];
// ColTupleList is NOT POD, so let's try this:
std::vector<WriteEngine::ColTupleList> colList;
// WriteEngine::dictStr dctColList[numCols];
std::vector<WriteEngine::dictStr> dctColList;
std::map<uint32_t, uint32_t> oids;
std::vector<BRM::OID_t> oidsToFlush;
// colpos = 0;
for (unsigned int ii = 0; ii < numCols; ii++)
{
colList.push_back(WriteEngine::ColTupleList());
dctColList.push_back(WriteEngine::dictStr());
}
try
{
DictOID dictOID = {0, 0, 0, 0, 0};
int dataType = convertDataType(colDefPtr->fType->fType);
if (dataType == CalpontSystemCatalog::DECIMAL || dataType == CalpontSystemCatalog::UDECIMAL)
{
if (colDefPtr->fType->fPrecision > 38) //@Bug 5717 precision cannot be over 38.
{
ostringstream os;
os << "Syntax error: The maximum precision (total number of digits) that can be specified is 38";
throw std::runtime_error(os.str());
}
else if (colDefPtr->fType->fPrecision < colDefPtr->fType->fScale)
{
ostringstream os;
os << "Syntax error: scale should be less than precision, precision: " << colDefPtr->fType->fPrecision
<< " scale: " << colDefPtr->fType->fScale;
throw std::runtime_error(os.str());
}
colDefPtr->convertDecimal();
}
if (dictoid > 0)
{
dictOID.compressionType = colDefPtr->fType->fCompressiontype;
dictOID.colWidth = colDefPtr->fType->fLength;
dictOID.dictOID = dictoid;
//@Bug 2534. Take away the limit of 255 and set the limit to 8000.
if ((colDefPtr->fType->fLength > 8000) && (dataType != CalpontSystemCatalog::BLOB) &&
(dataType != CalpontSystemCatalog::TEXT))
{
ostringstream os;
os << "char, varchar and varbinary length may not exceed 8000 bytes";
throw std::runtime_error(os.str());
}
}
else if ((dataType == CalpontSystemCatalog::VARBINARY || dataType == CalpontSystemCatalog::BLOB ||
dataType == CalpontSystemCatalog::TEXT) &&
colDefPtr->fType->fLength <= 7)
{
ostringstream os;
os << "varbinary and blob length may not be less than 8";
throw std::runtime_error(os.str());
}
unsigned int i = 0;
uint16_t dbRoot;
BRM::OID_t sysOid = 1021;
// Find out where syscolumn is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
column_iterator = columns.begin();
while (column_iterator != columns.end())
{
NullString tmpStr;
column = *column_iterator;
boost::to_lower(column.tableColName.column);
if (SCHEMA_COL == column.tableColName.column)
{
colTuple.data = schema;
tmpStr.assign(schema);
}
else if (TABLENAME_COL == column.tableColName.column)
{
colTuple.data = tablename;
tmpStr.assign(tablename);
}
else if (COLNAME_COL == column.tableColName.column)
{
boost::to_lower(colDefPtr->fName);
colTuple.data = colDefPtr->fName;
tmpStr.assign(colDefPtr->fName);
}
else if (OBJECTID_COL == column.tableColName.column)
{
colTuple.data = coloid;
}
else if (DATATYPE_COL == column.tableColName.column)
{
colTuple.data = dataType;
}
else if (COLUMNLEN_COL == column.tableColName.column)
{
//@Bug 2089 Disallow zero length char and varch column to be created
if (dataType == CalpontSystemCatalog::CHAR || dataType == CalpontSystemCatalog::VARCHAR ||
dataType == CalpontSystemCatalog::VARBINARY || dataType == CalpontSystemCatalog::BLOB ||
dataType == CalpontSystemCatalog::TEXT)
{
if (colDefPtr->fType->fLength <= 0)
{
ostringstream os;
os << "char, varchar and varbinary length must be greater than zero";
throw std::runtime_error(os.str());
}
}
colTuple.data = colDefPtr->fType->fLength;
}
else if (COLUMNPOS_COL == column.tableColName.column)
{
colTuple.data = startPos;
}
else if (DEFAULTVAL_COL == column.tableColName.column)
{
if (colDefPtr->fDefaultValue && !colDefPtr->fDefaultValue->fNull)
{
tmpStr.assign(colDefPtr->fDefaultValue->fValue);
}
else
{
tmpStr.dropString();
// colTuple.data = column.colType.getNullValueForType();
}
colTuple.data = tmpStr;
}
else if (NULLABLE_COL == column.tableColName.column)
{
int nullable = 1;
ColumnConstraintList& colConstraints = colDefPtr->fConstraints;
ColumnConstraintList::const_iterator constraint_iter = colConstraints.begin();
while (constraint_iter != colConstraints.end())
{
ColumnConstraintDef* consDefPtr = *constraint_iter;
if (consDefPtr->fConstraintType == ddlpackage::DDL_NOT_NULL)
{
nullable = 0;
break;
}
++constraint_iter;
}
colTuple.data = nullable;
}
else if (SCALE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fScale;
}
else if (PRECISION_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fPrecision;
}
else if (DICTOID_COL == column.tableColName.column)
{
if (dictoid > 0)
{
colTuple.data = dictOID.dictOID;
}
else
{
colTuple.data = column.colType.getNullValueForType();
}
}
else if (LISTOBJID_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (TREEOBJID_COL == column.tableColName.column)
{
colTuple.data = column.colType.getNullValueForType();
}
else if (MINVAL_COL == column.tableColName.column)
{
tmpStr.dropString();
}
else if (MAXVAL_COL == column.tableColName.column)
{
tmpStr.dropString();
}
else if (COMPRESSIONTYPE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fCompressiontype;
}
else if (AUTOINC_COL == column.tableColName.column)
{
// cout << "autoincrement= " << colDefPtr->fType->fAutoincrement << endl;
colTuple.data = colDefPtr->fType->fAutoincrement;
}
else if (NEXTVALUE_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fNextvalue;
}
else if (CHARSETNUM_COL == column.tableColName.column)
{
colTuple.data = colDefPtr->fType->fCharsetNum;
}
else
{
colTuple.data = column.colType.getNullValueForType();
}
colStruct.dataOid = column.oid;
oids[column.oid] = column.oid;
oidsToFlush.push_back(column.oid);
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.fColDbRoot = dbRoot;
colStruct.tokenFlag = column.colType.colWidth > 8 ? true : false;
colStruct.colDataType = column.colType.colDataType;
dctnryStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = column.oid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = column.oid;
}
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
cscColTypeList.push_back(column.colType);
colList[i].push_back(colTuple);
// colList.push_back(WriteEngine::ColTupleList());
// colList.back().push_back(colTuple);
dctColList[i].push_back(tmpStr);
// dctColList.push_back(WriteEngine::dictStr());
// dctColList.back().push_back(tmpStr);
++i;
++column_iterator;
}
if (0 != colStructs.size())
{
// FIXME: Is there a cleaner way to do this? Isn't colValuesList the same as colList after this?
for (unsigned int n = 0; n < numCols; n++)
{
colValuesList.push_back(colList[n]);
dctnryValueList.push_back(dctColList[n]);
}
// fWEWrapper.setDebugLevel(WriteEngine::DEBUG_3);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
int rc1 = 0;
error = fWEWrapper.insertColumnRec_SYS(txnID, cscColTypeList, colStructs, colValuesList,
dctnryStructList, dctnryValueList, SYSCOLUMN_BASE);
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error != WriteEngine::NO_ERROR)
{
if (error == ERR_BRM_WR_VB_ENTRY)
{
throw std::runtime_error("writeSysColumnMetaData WE: Error writing to BRM.");
}
else
{
WErrorCodes ec;
throw std::runtime_error("WE: Error updating calpontsys.syscolumn. " + ec.errorString(error));
}
}
else
error = rc1;
}
}
catch (exception& ex)
{
err += ex.what();
rc = 1;
}
catch (...)
{
err += "Unknown exception caught";
rc = 1;
}
purgeFDCache();
if (isAlter)
{
if (idbdatafile::IDBPolicy::useHdfs())
cacheutils::flushOIDsFromCache(oidsToFlush);
}
return rc;
}
uint8_t WE_DDLCommandProc::createtablefiles(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t size, i;
uint16_t tmp16;
uint32_t tmp32;
uint8_t tmp8;
OID dataOid;
int colWidth;
bool tokenFlag;
int txnID;
CalpontSystemCatalog::ColDataType colDataType;
uint16_t colDbRoot;
int compressionType;
bs >> tmp32;
txnID = tmp32;
bs >> size;
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(true);
std::map<uint32_t, uint32_t> oids;
for (i = 0; i < size; ++i)
{
bs >> tmp32;
dataOid = tmp32;
bs >> tmp8;
colDataType = (CalpontSystemCatalog::ColDataType)tmp8;
bs >> tmp8;
tokenFlag = (tmp8 != 0);
bs >> tmp32;
colWidth = tmp32;
bs >> tmp16;
colDbRoot = tmp16;
bs >> tmp32;
compressionType = tmp32;
oids[dataOid] = dataOid;
if (tokenFlag)
{
rc = fWEWrapper.createDctnry(0, dataOid, colWidth, colDbRoot, 0, 0, compressionType);
}
else
{
rc = fWEWrapper.createColumn(0, dataOid, colDataType, colWidth, colDbRoot, 0, compressionType);
}
if (rc != 0)
break;
}
// cout << "creating column file got error code " << rc << endl;
if (rc != 0)
{
WErrorCodes ec;
ostringstream oss;
oss << "WE: Error creating column file for oid " << dataOid << "; " << ec.errorString(rc) << endl;
err = oss.str();
}
// if (idbdatafile::IDBPolicy::useHdfs())
fWEWrapper.flushDataFiles(rc, txnID, oids);
purgeFDCache();
return rc;
}
uint8_t WE_DDLCommandProc::commitVersion(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tmp32;
int txnID;
bs >> tmp32;
txnID = tmp32;
rc = fWEWrapper.commit(txnID);
if (rc != 0)
{
WErrorCodes ec;
ostringstream oss;
oss << "WE: Error commiting transaction; " << txnID << ec.errorString(rc) << endl;
err = oss.str();
}
return rc;
}
uint8_t WE_DDLCommandProc::rollbackBlocks(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
;
int txnID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(true);
rc = fWEWrapper.rollbackBlocks(txnID, sessionID);
if (rc != 0)
{
WErrorCodes ec;
ostringstream oss;
oss << "WE: Error rolling back files " << txnID << " for session " << sessionID << "; "
<< ec.errorString(rc) << endl;
err = oss.str();
}
std::map<uint32_t, uint32_t> oids;
if (idbdatafile::IDBPolicy::useHdfs())
fWEWrapper.flushDataFiles(rc, txnID, oids);
purgeFDCache();
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
return rc;
}
uint8_t WE_DDLCommandProc::rollbackVersion(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
int txnID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
rc = fWEWrapper.rollbackVersion(txnID, sessionID);
if (rc != 0)
{
WErrorCodes ec;
ostringstream oss;
oss << "WE: Error rolling back transaction " << txnID << " for session " << sessionID << "; "
<< ec.errorString(rc) << endl;
err = oss.str();
}
purgeFDCache();
return rc;
}
uint8_t WE_DDLCommandProc::deleteSyscolumn(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
int txnID;
string schema, tablename;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
ddlpackage::QualifiedName sysCatalogTableName;
sysCatalogTableName.fSchema = CALPONT_SCHEMA;
sysCatalogTableName.fName = SYSCOLUMN_TABLE;
CalpontSystemCatalog::TableName userTableName;
userTableName.schema = schema;
userTableName.table = tablename;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
uint16_t dbRoot;
BRM::OID_t sysOid = 1021;
// Find out where syscolumn is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
fWEWrapper.setTransId(txnID);
fWEWrapper.startTransaction(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
try
{
CalpontSystemCatalog::RIDList colRidList = systemCatalogPtr->columnRIDs(userTableName);
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<void*> colValuesList;
WriteEngine::RIDList ridList;
std::vector<WriteEngine::RIDList> ridLists;
DDLColumn column;
CalpontSystemCatalog::RIDList::const_iterator colrid_iterator = colRidList.begin();
while (colrid_iterator != colRidList.end())
{
WriteEngine::RID rid = (*colrid_iterator).rid;
ridList.push_back(rid);
++colrid_iterator;
}
ColumnList columns;
getColumnsForTable(sessionID, sysCatalogTableName.fSchema, sysCatalogTableName.fName, columns);
ColumnList::const_iterator column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
++column_iterator;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
ridLists.push_back(ridList);
if (0 != colStructs.size() && 0 != ridLists[0].size())
{
int error = fWEWrapper.deleteRow(txnID, colExtentsColType, colExtentsStruct, colValuesList, ridLists,
SYSCOLUMN_BASE);
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error == NO_ERROR)
rc = rc1;
else
rc = error;
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
}
catch (...)
{
err = "Unknown exception caught";
rc = 1;
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::deleteSyscolumnRow(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
;
int txnID;
string schema, tablename, columnname;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
bs >> columnname;
ddlpackage::QualifiedName sysCatalogTableName;
sysCatalogTableName.fSchema = CALPONT_SCHEMA;
sysCatalogTableName.fName = SYSCOLUMN_TABLE;
CalpontSystemCatalog::TableColName tableColName;
tableColName.schema = schema;
tableColName.table = tablename;
tableColName.column = columnname;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
uint16_t dbRoot;
BRM::OID_t sysOid = 1021;
// Find out where syscolumn is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
try
{
CalpontSystemCatalog::ROPair colRO = systemCatalogPtr->columnRID(tableColName);
if (colRO.objnum < 0)
{
err = "Column not found:" + tableColName.table + "." + tableColName.column;
throw std::runtime_error(err);
}
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<void*> colValuesList;
WriteEngine::RIDList ridList;
std::vector<WriteEngine::RIDList> ridLists;
DDLColumn column;
ridList.push_back(colRO.rid);
ColumnList columns;
getColumnsForTable(sessionID, sysCatalogTableName.fSchema, sysCatalogTableName.fName, columns);
ColumnList::const_iterator column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
++column_iterator;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
ridLists.push_back(ridList);
if (0 != colStructs.size() && 0 != ridLists[0].size())
{
int error = fWEWrapper.deleteRow(txnID, colExtentsColType, colExtentsStruct, colValuesList, ridLists,
SYSCOLUMN_BASE);
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error == NO_ERROR)
rc = rc1;
else
rc = error;
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
}
catch (...)
{
err = "Unknown exception caught";
rc = 1;
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::deleteSystable(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
;
int txnID;
string schema, tablename;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
WriteEngine::WriteEngineWrapper writeEngine;
ddlpackage::QualifiedName sysCatalogTableName;
sysCatalogTableName.fSchema = CALPONT_SCHEMA;
sysCatalogTableName.fName = SYSTABLE_TABLE;
CalpontSystemCatalog::TableName userTableName;
userTableName.schema = schema;
userTableName.table = tablename;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
uint16_t dbRoot;
BRM::OID_t sysOid = 1001;
// Find out where systcolumn is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
fWEWrapper.setTransId(txnID);
fWEWrapper.startTransaction(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
try
{
CalpontSystemCatalog::ROPair userTableROPair = systemCatalogPtr->tableRID(userTableName);
if (userTableROPair.rid == std::numeric_limits<WriteEngine::RID>::max())
{
err = "RowID is not valid ";
throw std::runtime_error(err);
}
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<void*> colValuesList;
WriteEngine::RIDList ridList;
std::vector<WriteEngine::RIDList> ridLists;
DDLColumn column;
ridList.push_back(userTableROPair.rid);
ColumnList columns;
getColumnsForTable(sessionID, sysCatalogTableName.fSchema, sysCatalogTableName.fName, columns);
ColumnList::const_iterator column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
++column_iterator;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
ridLists.push_back(ridList);
if (0 != colStructs.size() && 0 != ridLists[0].size())
{
int error = fWEWrapper.deleteRow(txnID, colExtentsColType, colExtentsStruct, colValuesList, ridLists,
SYSCOLUMN_BASE);
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error == NO_ERROR)
rc = rc1;
else
rc = error;
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
}
catch (...)
{
err = "Unknown exception caught";
rc = 1;
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::deleteSystables(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
;
int txnID;
string schema, tablename;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
WriteEngine::WriteEngineWrapper writeEngine;
ddlpackage::QualifiedName sysCatalogTableName;
sysCatalogTableName.fSchema = CALPONT_SCHEMA;
sysCatalogTableName.fName = SYSTABLE_TABLE;
CalpontSystemCatalog::TableName userTableName;
userTableName.schema = schema;
userTableName.table = tablename;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
WriteEngine::ColStruct colStruct;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<void*> colValuesList;
WriteEngine::RIDList ridList;
std::vector<WriteEngine::RIDList> ridLists;
DDLColumn column;
uint16_t dbRoot;
BRM::OID_t sysOid = 1003;
// Find out where systable is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
try
{
CalpontSystemCatalog::ROPair userTableROPair = systemCatalogPtr->tableRID(userTableName);
if (userTableROPair.rid == std::numeric_limits<WriteEngine::RID>::max())
{
err = "RowID is not valid ";
throw std::runtime_error(err);
}
ridList.push_back(userTableROPair.rid);
ColumnList columns;
getColumnsForTable(sessionID, sysCatalogTableName.fSchema, sysCatalogTableName.fName, columns);
ColumnList::const_iterator column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
++column_iterator;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
ridLists.push_back(ridList);
{
int error = fWEWrapper.deleteRow(txnID, colExtentsColType, colExtentsStruct, colValuesList, ridLists,
SYSCOLUMN_BASE);
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error == NO_ERROR)
rc = rc1;
else
rc = error;
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
}
catch (...)
{
err = "Unknown exception caught";
rc = 1;
}
if (rc != 0)
return rc;
// deleting from SYSCOLUMN
sysCatalogTableName.fSchema = CALPONT_SCHEMA;
sysCatalogTableName.fName = SYSCOLUMN_TABLE;
sysOid = 1021;
// Find out where syscolumn is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
try
{
CalpontSystemCatalog::RIDList colRidList = systemCatalogPtr->columnRIDs(userTableName);
colStructs.clear();
cscColTypeList.clear();
colExtentsStruct.clear();
colExtentsColType.clear();
colValuesList.clear();
ridList.clear();
ridLists.clear();
oids.clear();
DDLColumn column;
CalpontSystemCatalog::RIDList::const_iterator colrid_iterator = colRidList.begin();
while (colrid_iterator != colRidList.end())
{
WriteEngine::RID rid = (*colrid_iterator).rid;
ridList.push_back(rid);
++colrid_iterator;
}
ColumnList columns;
getColumnsForTable(sessionID, sysCatalogTableName.fSchema, sysCatalogTableName.fName, columns);
ColumnList::const_iterator column_iterator = columns.begin();
while (column_iterator != columns.end())
{
column = *column_iterator;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.colDataType = column.colType.colDataType;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column.colType);
// oidsToFlush.push_back(colStruct.dataOid);
++column_iterator;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
ridLists.push_back(ridList);
if (0 != colStructs.size() && 0 != ridLists[0].size())
{
int error = fWEWrapper.deleteRow(txnID, colExtentsColType, colExtentsStruct, colValuesList, ridLists,
SYSCOLUMN_BASE);
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(error, txnID, oids);
if ((error == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (error == NO_ERROR)
rc = rc1;
else
rc = error;
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
}
catch (...)
{
err = "Unknown exception caught";
rc = 1;
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::dropFiles(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t size, i;
uint32_t tmp32;
std::vector<int32_t> dataOids;
bs >> size;
for (i = 0; i < size; ++i)
{
bs >> tmp32;
dataOids.push_back(tmp32);
}
try
{
rc = fWEWrapper.dropFiles(0, dataOids);
}
catch (...)
{
err = "WE: Error removing files ";
rc = 1;
}
purgeFDCache();
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnAuto(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
std::string schema, tablename;
int txnID;
uint8_t tmp8;
bool autoIncrement = false;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
bs >> tmp8;
autoIncrement = true;
CalpontSystemCatalog::TableName tableName;
tableName.schema = schema;
tableName.table = tablename;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
CalpontSystemCatalog::RIDList roList;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
try
{
roList = systemCatalogPtr->columnRIDs(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
// Build colStructs for SYSTABLE
std::vector<WriteEngine::RID> ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSCOLUMN_TABLE;
DDLColumn column;
WriteEngine::ColTuple colTuple;
findColumnData(sessionID, tableName, AUTOINC_COL, column);
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
string s1("y"), s2("n");
boost::any datavalue1 = s1;
boost::any datavalue2 = s2;
if (autoIncrement)
colTuple.data = datavalue1;
else
colTuple.data = datavalue2;
colStruct.colDataType = column.colType.colDataType;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
dctnryStructList.push_back(dctnryStruct);
cscColTypeList.push_back(column.colType);
for (unsigned int i = 0; i < roList.size(); i++)
{
aColList.push_back(colTuple);
}
colValuesList.push_back(aColList);
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
CalpontSystemCatalog::OID oid = 1021;
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.setTransId(txnID);
for (unsigned int i = 0; i < roList.size(); i++)
{
convertRidToColumn(roList[i].rid, dbRoot, partition, segment, oid);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
if (extentsinfo.empty())
extentsinfo.push_back(aExtentinfo);
else if (extentsinfo.back() != aExtentinfo)
extentsinfo.push_back(aExtentinfo);
ridList.push_back(roList[i].rid);
}
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
if (idbdatafile::IDBPolicy::useHdfs())
fWEWrapper.startTransaction(txnID);
rc = fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList, colOldValuesList,
ridLists, dctnryExtentsStruct, dctnryValueList, SYSCOLUMN_BASE);
if (rc != NO_ERROR)
{
// build the logging message
err = "WE: Update failed on: " + tableName.table;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnNextvalCol(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32;
std::string schema, tablename;
int txnID;
uint8_t tmp8;
bool autoIncrement = false;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
bs >> tmp8;
autoIncrement = true;
CalpontSystemCatalog::TableName tableName;
tableName.schema = schema;
tableName.table = tablename;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
CalpontSystemCatalog::RIDList roList;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
try
{
roList = systemCatalogPtr->columnRIDs(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
// Build colStructs for SYSTABLE
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSCOLUMN_TABLE;
DDLColumn column;
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::ColTuple colTuple;
std::vector<WriteEngine::RID> ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
boost::any datavalue;
findColumnData(sessionID, tableName, AUTOINC_COL, column);
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
string ystr("y");
string nstr("n");
if (autoIncrement)
colTuple.data = ystr;
else
colTuple.data = nstr;
colStruct.colDataType = column.colType.colDataType;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
cscColTypeList.push_back(column.colType);
for (unsigned int i = 0; i < roList.size(); i++)
{
aColList.push_back(colTuple);
}
colValuesList.push_back(aColList);
// get start dbroot for this PM.
// int PMNum = Config::getLocalModuleID();
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
// oam.getDbroots(PMNum);
// dbRoot will be the first dbroot on this pm. dbrootCnt will be how many dbroots on this PM.
CalpontSystemCatalog::OID oid = 1021;
for (unsigned int i = 0; i < roList.size(); i++)
{
convertRidToColumn(roList[i].rid, dbRoot, partition, segment, oid);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
if (extentsinfo.empty())
extentsinfo.push_back(aExtentinfo);
else if (extentsinfo.back() != aExtentinfo)
extentsinfo.push_back(aExtentinfo);
ridList.push_back(roList[i].rid);
}
std::vector<WriteEngine::RIDList> ridLists;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
ridLists.push_back(ridList);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
fWEWrapper.setTransId(txnID);
fWEWrapper.startTransaction(txnID);
rc = fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList, colOldValuesList,
ridLists, dctnryExtentsStruct, dctnryValueList, SYSCOLUMN_BASE);
if (rc != NO_ERROR)
{
// build the logging message
err = "WE: Update failed on: " + tableName.table;
}
if (idbdatafile::IDBPolicy::useHdfs())
{
fWEWrapper.flushDataFiles(rc, txnID, oids);
fWEWrapper.confirmTransaction(txnID);
if (rc == 0)
fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
systemCatalogPtr->flushCache();
purgeFDCache();
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
return rc;
}
uint8_t WE_DDLCommandProc::updateSystableEntryForSysColumn(int32_t /*sessionID*/, uint32_t txnID,
const DDLColumn& column, const std::string& value,
const std::string& /*oldValue*/,
execplan::CalpontSystemCatalog::RIDList& roList,
std::string& err)
{
int rc = 0;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
WriteEngine::ColTuple colTuple;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
std::vector<WriteEngine::RID> ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::DctnryTuple dictTuple;
dictTuple.isNull = false;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
if ((column.colType.colDataType == CalpontSystemCatalog::CHAR && column.colType.colWidth > 8) ||
(column.colType.colDataType == CalpontSystemCatalog::VARCHAR && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::VARBINARY && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::BLOB && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::TEXT && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::DECIMAL && column.colType.precision > 18) ||
(column.colType.colDataType == CalpontSystemCatalog::UDECIMAL &&
column.colType.precision > 18)) // token
{
colStruct.colWidth = 8;
colStruct.tokenFlag = true;
}
else
{
colStruct.colWidth = column.colType.colWidth;
}
colStruct.colDataType = column.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = colStruct.dataOid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
}
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
cscColTypeList.push_back(column.colType);
for (unsigned int i = 0; i < roList.size(); i++)
{
aColList.push_back(colTuple);
}
colValuesList.push_back(aColList);
// It's the same string for each column, so we just need one dictionary struct
void* dictTuplePtr = static_cast<void*>(&dictTuple);
memset(dictTuplePtr, 0, sizeof(dictTuple));
dictTuple.sigValue = (unsigned char*)value.c_str();
dictTuple.sigSize = value.length();
dictTuple.isNull = false;
dctColList = dictTuple;
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
CalpontSystemCatalog::OID oid = 1021;
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
for (unsigned int i = 0; i < roList.size(); i++)
{
convertRidToColumn(roList[i].rid, dbRoot, partition, segment, oid);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
if (extentsinfo.empty())
extentsinfo.push_back(aExtentinfo);
else if (extentsinfo.back() != aExtentinfo)
extentsinfo.push_back(aExtentinfo);
ridList.push_back(roList[i].rid);
}
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
rc = fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList, colOldValuesList,
ridLists, dctnryExtentsStruct, dctnryValueList, SYSCOLUMN_BASE);
if (rc != NO_ERROR)
{
// build the logging message
err = "WE: Update failed on: " + value;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnTablename(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, txnID;
std::string schema, oldTablename, newTablename, newSchema;
bs >> sessionID;
bs >> txnID;
bs >> schema;
bs >> oldTablename;
bs >> newTablename;
bs >> newSchema;
CalpontSystemCatalog::TableName tableName;
tableName.schema = schema;
tableName.table = oldTablename;
CalpontSystemCatalog::RIDList roList;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
try
{
roList = systemCatalogPtr->columnRIDs(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
// Build colStructs for SYSTABLE
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSCOLUMN_TABLE;
DDLColumn column;
findColumnData(sessionID, tableName, TABLENAME_COL, column);
rc = updateSystableEntryForSysColumn(sessionID, txnID, column, newTablename, oldTablename, roList, err);
if (newSchema != schema && rc == NO_ERROR)
{
findColumnData(sessionID, tableName, SCHEMA_COL, column);
rc = updateSystableEntryForSysColumn(sessionID, txnID, column, newSchema, schema, roList, err);
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
// cout << "rename:syscolumn is updated" << endl;
return rc;
}
uint8_t WE_DDLCommandProc::updateSystableAuto(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t sessionID, tmp32, autoVal;
std::string schema, tablename;
int txnID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tablename;
bs >> autoVal;
CalpontSystemCatalog::TableName tableName;
tableName.schema = schema;
tableName.table = tablename;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
CalpontSystemCatalog::ROPair ropair;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
try
{
ropair = systemCatalogPtr->tableRID(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
if (ropair.objnum < 0)
{
err = "No such table: " + tableName.table;
rc = 1;
return rc;
}
// now we have to prepare the various structures for the WE to update the column.
std::vector<WriteEngine::RID> ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
boost::any datavalue;
datavalue = autoVal;
WriteEngine::ColTuple colTuple;
// Build colStructs for SYSTABLE
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSTABLE_TABLE;
DDLColumn column;
findColumnData(sessionID, tableName, AUTOINC_COL, column);
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column.colType.colDataType;
colTuple.data = datavalue;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
colStructs.push_back(colStruct);
cscColTypeList.push_back(column.colType);
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
dctnryStructList.push_back(dctnryStruct);
aColList.push_back(colTuple);
colValuesList.push_back(aColList);
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
WriteEngine::DctnryTuple dctnryTuple;
dctColList = dctnryTuple;
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// In this case, there's only 1 row, so only one one extent, but keep it generic...
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
CalpontSystemCatalog::OID oid = 1003;
convertRidToColumn(ropair.rid, dbRoot, partition, segment, oid);
ridList.push_back(ropair.rid);
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
extentsinfo.push_back(aExtentinfo);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
colExtentsColType.push_back(cscColTypeList);
dctnryExtentsStruct.push_back(dctnryStructList);
}
// call the write engine to update the row
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
rc = fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList, colOldValuesList,
ridLists, dctnryExtentsStruct, dctnryValueList, SYSCOLUMN_BASE);
if (rc != NO_ERROR)
{
// build the logging message
err = "WE: Update failed on: " + tableName.table;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::updateSystableEntryForSysTable(int32_t /*sessionID*/, uint32_t txnID,
const DDLColumn& column, const std::string& value,
const std::string& oldValue,
CalpontSystemCatalog::ROPair ropair,
std::string& err)
{
std::vector<WriteEngine::RID> ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
std::map<uint32_t, uint32_t> oids;
WriteEngine::ColTuple colTuple;
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = true;
colStruct.colDataType = column.colType.colDataType;
// Tokenize the data value
WriteEngine::DctnryStruct dictStruct;
dictStruct.dctnryOid = column.colType.ddn.dictOID;
dictStruct.columnOid = column.colType.columnOID;
WriteEngine::DctnryTuple dictTuple;
dictTuple.isNull = false;
dictTuple.sigValue = (unsigned char*)value.c_str();
dictTuple.sigSize = value.length();
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = colStruct.dataOid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
}
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
aColList.push_back(colTuple);
colValuesList.push_back(aColList);
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
dctColList = dictTuple;
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// In this case, there's only 1 row, so only one one extent, but keep it generic...
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
CalpontSystemCatalog::OID oid = 1003;
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
convertRidToColumn(ropair.rid, dbRoot, partition, segment, oid);
ridList.push_back(ropair.rid);
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
extentsinfo.push_back(aExtentinfo);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
int rc =
fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList, colOldValuesList,
ridLists, dctnryExtentsStruct, dctnryValueList, SYSCOLUMN_BASE);
if (rc != NO_ERROR)
{
// build the logging message
err = "WE: Update failed on: " + oldValue;
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
if (rc != 0)
return rc;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
return rc;
}
uint8_t WE_DDLCommandProc::updateSystableTablename(ByteStream& bs, std::string& err)
{
uint8_t rc;
uint32_t sessionID, tmp32, txnID;
std::string schema, oldTablename, newTablename, newSchema;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> oldTablename;
bs >> newTablename;
bs >> newSchema;
CalpontSystemCatalog::TableName tableName;
tableName.schema = schema;
tableName.table = oldTablename;
CalpontSystemCatalog::ROPair ropair;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
try
{
ropair = systemCatalogPtr->tableRID(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
if (ropair.objnum < 0)
{
err = "No such table: " + tableName.table;
return 1;
}
// now we have to prepare the various structures for the WE to update the column.
// Build colStructs for SYSTABLE
tableName.schema = CALPONT_SCHEMA;
tableName.table = SYSTABLE_TABLE;
DDLColumn column;
findColumnData(sessionID, tableName, TABLENAME_COL, column);
rc = updateSystableEntryForSysTable(sessionID, txnID, column, newTablename, oldTablename, ropair, err);
if (newSchema != schema && rc == NO_ERROR)
{
findColumnData(sessionID, tableName, SCHEMA_COL, column);
rc = updateSystableEntryForSysTable(sessionID, txnID, column, newSchema, schema, ropair, err);
}
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
// cout << "rename:syscolumn is updated" << endl;
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnColumnposCol(messageqcpp::ByteStream& bs, std::string& err)
{
int rc = 0;
int colPos;
string schema, atableName;
uint32_t sessionID, tmp32;
int txnID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> atableName;
bs >> tmp32;
colPos = tmp32;
WriteEngine::RIDList ridList;
WriteEngine::ColValueList colValuesList;
WriteEngine::ColValueList colOldValuesList;
CalpontSystemCatalog::TableName tableName;
tableName.table = atableName;
tableName.schema = schema;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr =
CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
CalpontSystemCatalog::RIDList rids;
try
{
rids = systemCatalogPtr->columnRIDs(tableName);
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
CalpontSystemCatalog::RIDList::const_iterator rid_iter = rids.begin();
boost::any value;
WriteEngine::ColTupleList colTuples;
CalpontSystemCatalog::ColType columnType;
CalpontSystemCatalog::ROPair colRO;
// cout << "colpos is " << colPos << endl;
try
{
while (rid_iter != rids.end())
{
// look up colType
colRO = *rid_iter;
columnType = systemCatalogPtr->colType(colRO.objnum);
if (columnType.colPosition < colPos)
{
++rid_iter;
continue;
}
ridList.push_back(colRO.rid);
value = columnType.colPosition - 1;
WriteEngine::ColTuple colTuple;
colTuple.data = value;
colTuples.push_back(colTuple);
++rid_iter;
}
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
colValuesList.push_back(colTuples);
uint16_t dbRoot;
BRM::OID_t sysOid = 1021;
// Find out where systable is
rc = fDbrm.getSysCatDBRoot(sysOid, dbRoot);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
if (colTuples.size() > 0)
{
WriteEngine::ColStructList colStructs;
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::CSCTypesList cscColTypeList;
CalpontSystemCatalog::ColType colType;
// Build column structure for COLUMNPOS_COL
colType.columnOID = colStruct.dataOid = OID_SYSCOLUMN_COLUMNPOS;
colType.colWidth = colStruct.colWidth = 4;
colStruct.tokenFlag = false;
colType.colDataType = colStruct.colDataType = CalpontSystemCatalog::INT;
colStruct.fColDbRoot = dbRoot;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
}
colStructs.push_back(colStruct);
cscColTypeList.push_back(colType);
oids[colStruct.dataOid] = colStruct.dataOid;
// oidsToFlush.push_back(colStruct.dataOid);
rc = fWEWrapper.updateColumnRecs(txnID, cscColTypeList, colStructs, colValuesList, ridList,
SYSCOLUMN_BASE);
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::fillNewColumn(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tmp32;
uint8_t tmp8;
int txnID;
OID dataOid, dictOid, refColOID;
CalpontSystemCatalog::ColDataType dataType, refColDataType;
bool autoincrement;
int dataWidth, scale, precision, compressionType, refColWidth, refCompressionType;
string defaultValStr;
ColTuple defaultVal;
long timeZone;
bs >> tmp32;
txnID = tmp32;
bs >> tmp32;
dataOid = tmp32;
bs >> tmp32;
dictOid = tmp32;
bs >> tmp8;
dataType = (CalpontSystemCatalog::ColDataType)tmp8;
bs >> tmp8;
autoincrement = (tmp8 != 0);
bs >> tmp32;
dataWidth = tmp32;
bs >> tmp32;
scale = tmp32;
bs >> tmp32;
precision = tmp32;
bs >> defaultValStr;
bs >> tmp8;
compressionType = tmp8;
bs >> tmp32;
refColOID = tmp32;
bs >> tmp8;
refColDataType = (CalpontSystemCatalog::ColDataType)tmp8;
bs >> tmp32;
refColWidth = tmp32;
bs >> tmp8;
refCompressionType = tmp8;
messageqcpp::ByteStream::octbyte timeZoneTemp;
bs >> timeZoneTemp;
timeZone = timeZoneTemp;
// Find the fill in value
bool isNULL = false;
if (defaultValStr == "")
isNULL = true;
CalpontSystemCatalog::ColType colType;
colType.colDataType = static_cast<CalpontSystemCatalog::ColDataType>(dataType);
colType.colWidth = dataWidth;
colType.scale = scale;
colType.precision = precision;
bool pushWarning = false;
defaultVal.data = colType.convertColumnData(defaultValStr, pushWarning, timeZone, isNULL, false, false);
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(true);
fWEWrapper.setBulkFlag(true);
std::map<uint32_t, uint32_t> oids;
oids[dataOid] = dataOid;
oids[refColOID] = refColOID;
rc = fWEWrapper.fillColumn(txnID, dataOid, colType, defaultVal, refColOID, refColDataType, refColWidth,
refCompressionType, isNULL, compressionType, defaultValStr, dictOid,
autoincrement);
if (rc != 0)
{
WErrorCodes ec;
err = ec.errorString(rc);
}
purgeFDCache();
return rc;
}
uint8_t WE_DDLCommandProc::writeTruncateLog(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tableOid, numOid, tmp32;
bs >> tableOid;
bs >> numOid;
std::vector<uint32_t> oids;
for (uint32_t i = 0; i < numOid; i++)
{
bs >> tmp32;
oids.push_back(tmp32);
}
string prefix;
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
err = "Need a valid DBRMRoot entry in Calpont configuation file";
rc = 1;
return rc;
}
uint64_t pos = prefix.find_last_of("/");
std::string DDLLogFileName;
if (pos != string::npos)
{
DDLLogFileName = prefix.substr(0, pos + 1); // Get the file path
}
else
{
err = "Cannot find the dbrm directory for the DDL log file";
rc = 1;
return rc;
}
std::ostringstream oss;
oss << tableOid;
DDLLogFileName += "DDL_TRUNCATETABLE_Log_" + oss.str();
boost::scoped_ptr<idbdatafile::IDBDataFile> DDLLogFile(IDBDataFile::open(
IDBPolicy::getType(DDLLogFileName.c_str(), IDBPolicy::WRITEENG), DDLLogFileName.c_str(), "w", 0));
if (!DDLLogFile)
{
err = "DDL truncate table log file cannot be created";
rc = 1;
return rc;
}
std::ostringstream buf;
for (unsigned i = 0; i < oids.size(); i++)
buf << oids[i] << std::endl;
std::string tmp(buf.str());
DDLLogFile->write(tmp.c_str(), tmp.size());
// DDLLogFile is a scoped_ptr, will be closed after return.
return rc;
}
uint8_t WE_DDLCommandProc::writeDropPartitionLog(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tableOid, numParts, numOid, tmp32;
bs >> tableOid;
std::set<BRM::LogicalPartition> partitionNums;
bs >> numParts;
BRM::LogicalPartition lp;
for (uint32_t i = 0; i < numParts; i++)
{
lp.unserialize(bs);
partitionNums.insert(lp);
}
bs >> numOid;
std::vector<uint32_t> oids;
for (uint32_t i = 0; i < numOid; i++)
{
bs >> tmp32;
oids.push_back(tmp32);
}
string prefix;
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
err = "Need a valid DBRMRoot entry in Calpont configuation file";
rc = 1;
return rc;
}
uint64_t pos = prefix.find_last_of("/");
std::string DDLLogFileName;
if (pos != string::npos)
{
DDLLogFileName = prefix.substr(0, pos + 1); // Get the file path
}
else
{
err = "Cannot find the dbrm directory for the DDL drop partitions log file";
rc = 1;
return rc;
}
std::ostringstream oss;
oss << tableOid;
DDLLogFileName += "DDL_DROPPARTITION_Log_" + oss.str();
boost::scoped_ptr<idbdatafile::IDBDataFile> DDLLogFile(IDBDataFile::open(
IDBPolicy::getType(DDLLogFileName.c_str(), IDBPolicy::WRITEENG), DDLLogFileName.c_str(), "w", 0));
if (!DDLLogFile)
{
err = "DDL drop partitions log file cannot be created";
rc = 1;
return rc;
}
std::ostringstream buf;
// @SN write partition numbers to the log file, separated by space
set<BRM::LogicalPartition>::const_iterator it;
for (it = partitionNums.begin(); it != partitionNums.end(); ++it)
buf << (*it) << endl;
// -1 indicates the end of partition list
BRM::LogicalPartition end(-1, -1, -1);
buf << end << endl;
for (unsigned i = 0; i < oids.size(); i++)
buf << oids[i] << std::endl;
std::string tmp(buf.str());
DDLLogFile->write(tmp.c_str(), tmp.size());
return rc;
}
uint8_t WE_DDLCommandProc::writeDropTableLog(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tableOid, numOid, tmp32;
bs >> tableOid;
bs >> numOid;
std::vector<uint32_t> oids;
for (uint32_t i = 0; i < numOid; i++)
{
bs >> tmp32;
oids.push_back(tmp32);
}
string prefix;
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
err = "Need a valid DBRMRoot entry in Calpont configuation file";
rc = 1;
return rc;
}
uint64_t pos = prefix.find_last_of("/");
std::string DDLLogFileName;
if (pos != string::npos)
{
DDLLogFileName = prefix.substr(0, pos + 1); // Get the file path
}
else
{
err = "Cannot find the dbrm directory for the DDL drop partitions log file";
rc = 1;
return rc;
}
std::ostringstream oss;
oss << tableOid;
DDLLogFileName += "DDL_DROPTABLE_Log_" + oss.str();
boost::scoped_ptr<idbdatafile::IDBDataFile> DDLLogFile(IDBDataFile::open(
IDBPolicy::getType(DDLLogFileName.c_str(), IDBPolicy::WRITEENG), DDLLogFileName.c_str(), "w", 0));
if (!DDLLogFile)
{
err = "DDL drop table log file cannot be created";
rc = 1;
return rc;
}
std::ostringstream buf;
for (unsigned i = 0; i < oids.size(); i++)
buf << oids[i] << std::endl;
std::string tmp(buf.str());
DDLLogFile->write(tmp.c_str(), tmp.size());
return rc;
}
uint8_t WE_DDLCommandProc::deleteDDLLog(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t tableOid, fileType;
bs >> fileType;
bs >> tableOid;
string prefix;
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
err = "Need a valid DBRMRoot entry in Calpont configuation file";
rc = 1;
return rc;
}
uint64_t pos = prefix.find_last_of("/");
std::string DDLLogFileName;
if (pos != string::npos)
{
DDLLogFileName = prefix.substr(0, pos + 1); // Get the file path
}
else
{
err = "Cannot find the dbrm directory for the DDL drop partitions log file";
rc = 1;
return rc;
}
std::ostringstream oss;
oss << tableOid;
switch (fileType)
{
case DROPTABLE_LOG:
{
DDLLogFileName += "DDL_DROPTABLE_Log_" + oss.str();
break;
}
case DROPPART_LOG:
{
DDLLogFileName += "DDL_DROPPARTITION_Log_" + oss.str();
break;
}
case TRUNCATE_LOG:
{
DDLLogFileName += "DDL_TRUNCATETABLE_Log_" + oss.str();
break;
}
default: break;
}
IDBPolicy::remove(DDLLogFileName.c_str());
return rc;
}
uint8_t WE_DDLCommandProc::fetchDDLLog(ByteStream& bs, std::string& err)
{
int rc = 0;
// Find the ddl log files under DBRMRoot directory
string prefix, ddlLogDir;
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
rc = 1;
err = "Need a valid DBRMRoot entry in Calpont configuation file";
return rc;
}
uint64_t pos = prefix.find_last_of("/");
if (pos != string::npos)
{
ddlLogDir = prefix.substr(0, pos + 1); // Get the file path
}
else
{
rc = 1;
err = "Cannot find the dbrm directory for the DDL log file";
return rc;
}
boost::filesystem::path filePath;
filePath = fs::system_complete(fs::path(ddlLogDir));
if (!fs::exists(filePath))
{
rc = 1;
err = "\nDDL log file path is Not found: ";
return rc;
}
std::vector<string> fileNames;
if (fs::is_directory(filePath))
{
fs::directory_iterator end_iter;
for (fs::directory_iterator dir_itr(filePath); dir_itr != end_iter; ++dir_itr)
{
try
{
if (!fs::is_directory(*dir_itr))
{
#if BOOST_VERSION >= 105200
fileNames.push_back(dir_itr->path().generic_string());
#else
fileNames.push_back(dir_itr->string());
#endif
}
}
catch (std::exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
}
}
CalpontSystemCatalog::OID fileoid;
string tableName;
bs.restart();
for (unsigned i = 0; i < fileNames.size(); i++)
{
pos = fileNames[i].find("DDL_DROPTABLE_Log_");
if (pos != string::npos)
{
// Read the file to get oids
// cout << "Found file " << fileNames[i] << endl;
boost::scoped_ptr<idbdatafile::IDBDataFile> ddlLogFile(IDBDataFile::open(
IDBPolicy::getType(fileNames[i].c_str(), IDBPolicy::WRITEENG), fileNames[i].c_str(), "r", 0));
if (!ddlLogFile)
continue;
// find the table oid
pos = fileNames[i].find_last_of("_");
string tableOidStr = fileNames[i].substr(pos + 1, fileNames[i].length() - pos - 1);
char* ep = NULL;
uint32_t tableOid = strtoll(tableOidStr.c_str(), &ep, 10);
bs << tableOid;
bs << (uint32_t)DROPTABLE_LOG;
std::vector<CalpontSystemCatalog::OID> oidList;
ssize_t fileSize = ddlLogFile->size();
boost::scoped_array<char> buf(new char[fileSize]);
if (ddlLogFile->read(buf.get(), fileSize) != fileSize)
return (uint8_t)ERR_FILE_READ;
std::istringstream strbuf(string(buf.get(), fileSize));
while (strbuf >> fileoid)
oidList.push_back(fileoid);
bs << (uint32_t)oidList.size();
for (unsigned j = 0; j < oidList.size(); j++)
{
bs << (uint32_t)oidList[j];
}
bs << (uint32_t)0;
}
else // Find drop partition log file
{
pos = fileNames[i].find("DDL_DROPPARTITION_Log_");
if (pos != string::npos)
{
boost::scoped_ptr<idbdatafile::IDBDataFile> ddlLogFile(IDBDataFile::open(
IDBPolicy::getType(fileNames[i].c_str(), IDBPolicy::WRITEENG), fileNames[i].c_str(), "r", 0));
BRM::LogicalPartition partition;
vector<BRM::LogicalPartition> partitionNums;
// find the table oid
pos = fileNames[i].find_last_of("_");
string tableOidStr = fileNames[i].substr(pos + 1, fileNames[i].length() - pos - 1);
char* ep = NULL;
uint32_t tableOid = strtoll(tableOidStr.c_str(), &ep, 10);
bs << tableOid;
bs << (uint32_t)DROPPART_LOG;
ssize_t fileSize = ddlLogFile->size();
boost::scoped_array<char> buf(new char[fileSize]);
if (ddlLogFile->read(buf.get(), fileSize) != fileSize)
return (uint8_t)ERR_FILE_READ;
std::istringstream strbuf(string(buf.get(), fileSize));
while (strbuf >> partition)
{
if (partition.dbroot == (uint16_t)-1)
break;
partitionNums.push_back(partition);
}
std::vector<CalpontSystemCatalog::OID> oidPartList;
while (strbuf >> fileoid)
oidPartList.push_back(fileoid);
bs << (uint32_t)oidPartList.size();
for (unsigned j = 0; j < oidPartList.size(); j++)
{
bs << (uint32_t)oidPartList[j];
}
bs << (uint32_t)partitionNums.size();
for (unsigned j = 0; j < partitionNums.size(); j++)
{
partitionNums[j].serialize(bs);
}
}
else // find truncate table log file
{
pos = fileNames[i].find("DDL_TRUNCATETABLE_Log_");
if (pos != string::npos)
{
boost::scoped_ptr<idbdatafile::IDBDataFile> ddlLogFile(IDBDataFile::open(
IDBPolicy::getType(fileNames[i].c_str(), IDBPolicy::WRITEENG), fileNames[i].c_str(), "r", 0));
if (!ddlLogFile)
{
continue;
}
// find the table oid
pos = fileNames[i].find_last_of("_");
string tableOidStr = fileNames[i].substr(pos + 1, fileNames[i].length() - pos - 1);
char* ep = NULL;
uint32_t tableOid = strtoll(tableOidStr.c_str(), &ep, 10);
bs << tableOid;
bs << (uint32_t)TRUNCATE_LOG;
std::vector<CalpontSystemCatalog::OID> oidList;
ssize_t fileSize = ddlLogFile->size();
boost::scoped_array<char> buf(new char[fileSize]);
if (ddlLogFile->read(buf.get(), fileSize) != fileSize)
return (uint8_t)ERR_FILE_READ;
std::istringstream strbuf(string(buf.get(), fileSize));
while (strbuf >> fileoid)
oidList.push_back(fileoid);
bs << (uint32_t)oidList.size();
for (unsigned j = 0; j < oidList.size(); j++)
{
bs << (uint32_t)oidList[j];
}
bs << (uint32_t)0;
}
}
}
}
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnSetDefault(messageqcpp::ByteStream& bs, std::string& err)
{
// Will update five columns: columnname, defaultvalue, nullable, autoincrement, nextvalue.
int rc = 0;
uint32_t tmp32;
string schema, tableName, colName, defaultvalue;
int txnID;
uint32_t sessionID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tableName;
bs >> colName;
bs >> defaultvalue;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
CalpontSystemCatalog::TableName atableName;
CalpontSystemCatalog::TableColName tableColName;
tableColName.schema = schema;
tableColName.table = tableName;
tableColName.column = colName;
CalpontSystemCatalog::ROPair ropair;
try
{
ropair = systemCatalogPtr->columnRID(tableColName);
if (ropair.objnum < 0)
{
ostringstream oss;
oss << "No such column: " << tableColName;
throw std::runtime_error(oss.str().c_str());
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
catch (...)
{
err = "renameColumn:Unknown exception caught";
rc = 1;
return rc;
}
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
std::vector<WriteEngine::RID> ridList;
ridList.push_back(ropair.rid);
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList1;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
WriteEngine::ColTuple colTuple;
// Build colStructs for SYSCOLUMN
atableName.schema = CALPONT_SCHEMA;
atableName.table = SYSCOLUMN_TABLE;
DDLColumn column;
findColumnData(sessionID, atableName, DEFAULTVAL_COL, column); // DEFAULTVAL_COL column
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.setTransId(txnID);
fWEWrapper.startTransaction(txnID);
// Build DEFAULTVAL_COL structure
WriteEngine::ColTupleList aColList;
colStruct.dataOid = column.oid;
colStruct.colWidth = column.colType.colWidth > 8 ? 8 : column.colType.colWidth;
colStruct.tokenFlag = false;
if ((column.colType.colDataType == CalpontSystemCatalog::CHAR && column.colType.colWidth > 8) ||
(column.colType.colDataType == CalpontSystemCatalog::VARCHAR && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::VARBINARY && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::BLOB && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::TEXT && column.colType.colWidth > 7) ||
(column.colType.colDataType == CalpontSystemCatalog::DECIMAL && column.colType.precision > 18) ||
(column.colType.colDataType == CalpontSystemCatalog::UDECIMAL &&
column.colType.precision > 18)) // token
{
colStruct.colWidth = 8;
colStruct.tokenFlag = true;
}
else
{
colStruct.colWidth = column.colType.colWidth;
}
colStruct.colDataType = column.colType.colDataType;
if (colStruct.tokenFlag)
{
WriteEngine::DctnryStruct dictStruct;
dictStruct.dctnryOid = column.colType.ddn.dictOID;
dictStruct.columnOid = column.colType.columnOID;
if (defaultvalue.length() <= 0) // null token
{
WriteEngine::Token nullToken;
colTuple.data = nullToken;
}
else
{
WriteEngine::DctnryTuple dictTuple;
dictTuple.sigValue = (unsigned char*)defaultvalue.c_str();
dictTuple.sigSize = defaultvalue.length();
dictTuple.isNull = false;
int error = NO_ERROR;
if (NO_ERROR !=
(error = fWEWrapper.tokenize(txnID, dictStruct, dictTuple, false))) // @bug 5572 HDFS tmp file
{
WErrorCodes ec;
throw std::runtime_error("WE: Tokenization failed " + ec.errorString(error));
}
WriteEngine::Token aToken = dictTuple.token;
colTuple.data = aToken;
}
}
colStruct.colDataType = column.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column.colType.charsetNumber;
dctnryStruct.columnOid = colStruct.dataOid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
}
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
dctnryStructList.push_back(dctnryStruct);
aColList.push_back(colTuple);
colValuesList.push_back(aColList);
WriteEngine::DctnryTuple dctnryTuple;
if (defaultvalue.length() > 0)
{
dctnryTuple.sigValue = (unsigned char*)defaultvalue.c_str();
dctnryTuple.sigSize = defaultvalue.length();
dctnryTuple.isNull = false;
}
else
{
dctnryTuple.isNull = true;
}
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
// In this case, there's only 1 row, so only one one extent, but keep it generic...
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
convertRidToColumn(ropair.rid, dbRoot, partition, segment, 1021);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
extentsinfo.push_back(aExtentinfo);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
if (NO_ERROR != fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList,
colOldValuesList, ridLists, dctnryExtentsStruct, dctnryValueList,
SYSCOLUMN_BASE))
{
err = "WE: Update failed on: " + atableName.table;
rc = 1;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
// flush syscat cahche
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::updateSyscolumnRenameColumn(messageqcpp::ByteStream& bs, std::string& err)
{
// Will update five columns: columnname, defaultvalue, nullable, autoincrement, nextvalue.
int rc = 0;
uint64_t nextVal;
uint32_t tmp32, nullable;
string schema, tableName, colOldname, autoinc, colNewName, defaultvalue;
int txnID;
uint32_t sessionID;
bs >> sessionID;
bs >> tmp32;
txnID = tmp32;
bs >> schema;
bs >> tableName;
bs >> colOldname;
bs >> colNewName;
bs >> autoinc;
bs >> nextVal;
bs >> nullable;
bs >> defaultvalue;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
CalpontSystemCatalog::TableName atableName;
CalpontSystemCatalog::TableColName tableColName;
tableColName.schema = schema;
tableColName.table = tableName;
tableColName.column = colOldname;
CalpontSystemCatalog::ROPair ropair;
try
{
ropair = systemCatalogPtr->columnRID(tableColName);
if (ropair.objnum < 0)
{
ostringstream oss;
oss << "No such column: " << tableColName;
throw std::runtime_error(oss.str().c_str());
}
}
catch (exception& ex)
{
err = ex.what();
rc = 1;
return rc;
}
catch (...)
{
err = "renameColumn:Unknown exception caught";
rc = 1;
return rc;
}
uint16_t dbRoot = 0;
uint16_t segment;
uint32_t partition;
std::vector<WriteEngine::RID> ridList;
ridList.push_back(ropair.rid);
WriteEngine::ColValueList colValuesList;
WriteEngine::ColTupleList aColList1;
WriteEngine::ColStructList colStructs;
WriteEngine::CSCTypesList cscColTypeList;
std::vector<void*> colOldValuesList;
std::map<uint32_t, uint32_t> oids;
// std::vector<BRM::OID_t> oidsToFlush;
WriteEngine::DctnryStructList dctnryStructList;
WriteEngine::DctnryValueList dctnryValueList;
WriteEngine::DctColTupleList dctRowList;
WriteEngine::DctnryTuple dctColList;
boost::any datavalue;
datavalue = colNewName;
WriteEngine::ColTuple colTuple;
// Build colStructs for SYSCOLUMN
atableName.schema = CALPONT_SCHEMA;
atableName.table = SYSCOLUMN_TABLE;
DDLColumn column1, column2, column3, column4, column5;
findColumnData(sessionID, atableName, COLNAME_COL, column1); // COLNAME_COL column
findColumnData(sessionID, atableName, AUTOINC_COL, column2); // AUTOINC_COL column
findColumnData(sessionID, atableName, NEXTVALUE_COL, column3); // NEXTVALUE_COL column
findColumnData(sessionID, atableName, NULLABLE_COL, column4); // NULLABLE_COL column
findColumnData(sessionID, atableName, DEFAULTVAL_COL, column5); // DEFAULTVAL_COL column
WriteEngine::ColStruct colStruct;
WriteEngine::DctnryStruct dctnryStruct;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
fWEWrapper.setTransId(txnID);
fWEWrapper.setIsInsert(false);
fWEWrapper.setBulkFlag(false);
fWEWrapper.startTransaction(txnID);
// Build COLNAME_COL structure
colStruct.dataOid = column1.oid;
colStruct.colWidth = column1.colType.colWidth > 8 ? 8 : column1.colType.colWidth;
colStruct.tokenFlag = false;
if ((column1.colType.colDataType == CalpontSystemCatalog::CHAR && column1.colType.colWidth > 8) ||
(column1.colType.colDataType == CalpontSystemCatalog::VARCHAR && column1.colType.colWidth > 7) ||
(column1.colType.colDataType == CalpontSystemCatalog::VARBINARY && column1.colType.colWidth > 7) ||
(column1.colType.colDataType == CalpontSystemCatalog::BLOB && column1.colType.colWidth > 7) ||
(column1.colType.colDataType == CalpontSystemCatalog::TEXT && column1.colType.colWidth > 7) ||
(column1.colType.colDataType == CalpontSystemCatalog::DECIMAL && column1.colType.precision > 18) ||
(column1.colType.colDataType == CalpontSystemCatalog::UDECIMAL &&
column1.colType.precision > 18)) // token
{
colStruct.colWidth = 8;
colStruct.tokenFlag = true;
}
else
{
colStruct.colWidth = column1.colType.colWidth;
}
colStruct.colDataType = column1.colType.colDataType;
if (colStruct.tokenFlag)
{
WriteEngine::DctnryStruct dictStruct;
if (idbdatafile::IDBPolicy::useHdfs())
{
dictStruct.fCompressionType = 2;
}
dictStruct.dctnryOid = column1.colType.ddn.dictOID;
dictStruct.columnOid = column1.colType.columnOID;
WriteEngine::DctnryTuple dictTuple;
dictTuple.sigValue = (unsigned char*)colNewName.c_str();
dictTuple.sigSize = colNewName.length();
dictTuple.isNull = false;
int error = NO_ERROR;
if (NO_ERROR !=
(error = fWEWrapper.tokenize(txnID, dictStruct, dictTuple, false))) // @bug 5572 HDFS tmp file
{
WErrorCodes ec;
err = ec.errorString(error);
rc = error;
return rc;
}
WriteEngine::Token aToken = dictTuple.token;
colTuple.data = aToken;
}
else
colTuple.data = datavalue;
colStruct.colDataType = column1.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column1.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column1.colType.charsetNumber;
dctnryStruct.columnOid = colStruct.dataOid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
}
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column1.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
dctnryStructList.push_back(dctnryStruct);
aColList1.push_back(colTuple);
colValuesList.push_back(aColList1);
WriteEngine::DctnryTuple dctnryTuple;
boost::to_lower(colNewName);
dctnryTuple.sigValue = (unsigned char*)colNewName.c_str();
dctnryTuple.sigSize = colNewName.length();
dctnryTuple.isNull = false;
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// Build AUTOINC_COL structure
WriteEngine::ColTupleList aColList2;
colStruct.dataOid = column2.oid;
colStruct.colWidth = column2.colType.colWidth > 8 ? 8 : column2.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column2.colType.colDataType;
colTuple.data = autoinc;
colStruct.colDataType = column2.colType.colDataType;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column2.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
dctnryStructList.push_back(dctnryStruct);
aColList2.push_back(colTuple);
colValuesList.push_back(aColList2);
dctnryTuple.isNull = true;
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// Build NEXTVALUE_COL structure
WriteEngine::ColTupleList aColList3;
colStruct.dataOid = column3.oid;
colStruct.colWidth = column3.colType.colWidth > 8 ? 8 : column3.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column3.colType.colDataType;
colTuple.data = nextVal;
colStruct.colDataType = column3.colType.colDataType;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column3.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
dctnryStructList.push_back(dctnryStruct);
aColList3.push_back(colTuple);
colValuesList.push_back(aColList3);
dctnryTuple.isNull = true;
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// Build NULLABLE_COL structure
WriteEngine::ColTupleList aColList4;
colStruct.dataOid = column4.oid;
colStruct.colWidth = column4.colType.colWidth > 8 ? 8 : column4.colType.colWidth;
colStruct.tokenFlag = false;
colStruct.colDataType = column4.colType.colDataType;
colTuple.data = nullable;
colStruct.colDataType = column4.colType.colDataType;
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
colStructs.push_back(colStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column4.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
dctnryStructList.push_back(dctnryStruct);
aColList4.push_back(colTuple);
colValuesList.push_back(aColList4);
dctnryTuple.isNull = true;
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
// Build DEFAULTVAL_COL structure
WriteEngine::ColTupleList aColList5;
colStruct.dataOid = column5.oid;
colStruct.colWidth = column5.colType.colWidth > 8 ? 8 : column5.colType.colWidth;
colStruct.tokenFlag = false;
if ((column5.colType.colDataType == CalpontSystemCatalog::CHAR && column5.colType.colWidth > 8) ||
(column5.colType.colDataType == CalpontSystemCatalog::VARCHAR && column5.colType.colWidth > 7) ||
(column5.colType.colDataType == CalpontSystemCatalog::VARBINARY && column5.colType.colWidth > 7) ||
(column5.colType.colDataType == CalpontSystemCatalog::BLOB && column5.colType.colWidth > 7) ||
(column5.colType.colDataType == CalpontSystemCatalog::TEXT && column5.colType.colWidth > 7) ||
(column5.colType.colDataType == CalpontSystemCatalog::DECIMAL && column5.colType.precision > 18) ||
(column5.colType.colDataType == CalpontSystemCatalog::UDECIMAL &&
column5.colType.precision > 18)) // token
{
colStruct.colWidth = 8;
colStruct.tokenFlag = true;
}
else
{
colStruct.colWidth = column5.colType.colWidth;
}
colStruct.colDataType = column5.colType.colDataType;
if (colStruct.tokenFlag)
{
WriteEngine::DctnryStruct dictStruct;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dictStruct.fCompressionType = 2;
}
dictStruct.dctnryOid = column5.colType.ddn.dictOID;
dictStruct.columnOid = column5.colType.columnOID;
if (defaultvalue.length() <= 0) // null token
{
WriteEngine::Token nullToken;
colTuple.data = nullToken;
}
else
{
WriteEngine::DctnryTuple dictTuple;
dictTuple.sigValue = (unsigned char*)defaultvalue.c_str();
dictTuple.sigSize = defaultvalue.length();
dictTuple.isNull = false;
int error = NO_ERROR;
if (NO_ERROR !=
(error = fWEWrapper.tokenize(txnID, dictStruct, dictTuple, false))) // @bug 5572 HDFS tmp file
{
WErrorCodes ec;
throw std::runtime_error("WE: Tokenization failed " + ec.errorString(error));
}
WriteEngine::Token aToken = dictTuple.token;
colTuple.data = aToken;
}
}
fWEWrapper.flushDataFiles(rc, txnID, oids);
colStruct.colDataType = column5.colType.colDataType;
if (idbdatafile::IDBPolicy::useHdfs())
{
colStruct.fCompressionType = 2;
dctnryStruct.fCompressionType = 2;
}
if (colStruct.tokenFlag)
{
dctnryStruct.dctnryOid = column5.colType.ddn.dictOID;
dctnryStruct.fCharsetNumber = column5.colType.charsetNumber;
dctnryStruct.columnOid = colStruct.dataOid;
}
else
{
dctnryStruct.dctnryOid = 0;
dctnryStruct.columnOid = colStruct.dataOid;
}
colStructs.push_back(colStruct);
dctnryStructList.push_back(dctnryStruct);
oids[colStruct.dataOid] = colStruct.dataOid;
cscColTypeList.push_back(column5.colType);
// oidsToFlush.push_back(colStruct.dataOid);
if (dctnryStruct.dctnryOid > 0)
{
oids[dctnryStruct.dctnryOid] = dctnryStruct.dctnryOid;
// oidsToFlush.push_back(dctnryStruct.dctnryOid);
}
aColList5.push_back(colTuple);
colValuesList.push_back(aColList5);
if (defaultvalue.length() > 0)
{
dctnryTuple.sigValue = (unsigned char*)defaultvalue.c_str();
dctnryTuple.sigSize = defaultvalue.length();
dctnryTuple.isNull = false;
}
else
{
dctnryTuple.isNull = true;
}
dctColList = dctnryTuple;
dctRowList.clear();
dctRowList.push_back(dctColList);
dctnryValueList.push_back(dctRowList);
std::vector<WriteEngine::ColStructList> colExtentsStruct;
std::vector<WriteEngine::CSCTypesList> colExtentsColType;
std::vector<WriteEngine::DctnryStructList> dctnryExtentsStruct;
std::vector<WriteEngine::RIDList> ridLists;
ridLists.push_back(ridList);
// In this case, there's only 1 row, so only one one extent, but keep it generic...
std::vector<extentInfo> extentsinfo;
extentInfo aExtentinfo;
convertRidToColumn(ropair.rid, dbRoot, partition, segment, 1021);
aExtentinfo.dbRoot = dbRoot;
aExtentinfo.partition = partition;
aExtentinfo.segment = segment;
extentsinfo.push_back(aExtentinfo);
// build colExtentsStruct
for (unsigned i = 0; i < extentsinfo.size(); i++)
{
for (unsigned j = 0; j < colStructs.size(); j++)
{
colStructs[j].fColPartition = extentsinfo[i].partition;
colStructs[j].fColSegment = extentsinfo[i].segment;
colStructs[j].fColDbRoot = extentsinfo[i].dbRoot;
dctnryStructList[j].fColPartition = extentsinfo[i].partition;
dctnryStructList[j].fColSegment = extentsinfo[i].segment;
dctnryStructList[j].fColDbRoot = extentsinfo[i].dbRoot;
}
colExtentsStruct.push_back(colStructs);
dctnryExtentsStruct.push_back(dctnryStructList);
colExtentsColType.push_back(cscColTypeList);
}
// call the write engine to update the row
if (NO_ERROR != fWEWrapper.updateColumnRec(txnID, colExtentsColType, colExtentsStruct, colValuesList,
colOldValuesList, ridLists, dctnryExtentsStruct, dctnryValueList,
SYSCOLUMN_BASE))
{
err = "WE: Update failed on: " + atableName.table;
rc = 1;
}
int rc1 = 0;
if (idbdatafile::IDBPolicy::useHdfs())
{
rc1 = fWEWrapper.flushDataFiles(rc, txnID, oids);
if ((rc == 0) && (rc1 == 0))
{
rc1 = fWEWrapper.confirmTransaction(txnID);
if (rc1 == NO_ERROR)
rc1 = fWEWrapper.endTransaction(txnID, true);
else
fWEWrapper.endTransaction(txnID, false);
}
else
{
fWEWrapper.endTransaction(txnID, false);
}
}
if (rc == 0)
rc = rc1;
// flush syscat cahche
systemCatalogPtr->flushCache();
purgeFDCache();
// if (idbdatafile::IDBPolicy::useHdfs())
// cacheutils::flushOIDsFromCache(oidsToFlush);
return rc;
}
uint8_t WE_DDLCommandProc::dropPartitions(ByteStream& bs, std::string& err)
{
int rc = 0;
uint32_t size, i;
uint32_t tmp32;
std::vector<OID> dataOids;
std::vector<BRM::PartitionInfo> partitions;
bs >> size;
for (i = 0; i < size; ++i)
{
bs >> tmp32;
dataOids.push_back(tmp32);
}
bs >> size;
BRM::PartitionInfo pi;
for (i = 0; i < size; ++i)
{
pi.unserialize(bs);
partitions.push_back(pi);
}
try
{
rc = fWEWrapper.deletePartitions(dataOids, partitions);
}
catch (...)
{
err = "WE: Error removing files ";
rc = 1;
}
return rc;
}
void WE_DDLCommandProc::purgeFDCache()
{
if (idbdatafile::IDBPolicy::useHdfs())
{
TableMetaData* aTbaleMetaData = TableMetaData::makeTableMetaData(SYSCOLUMN_BASE);
ColsExtsInfoMap colsExtsInfoMap = aTbaleMetaData->getColsExtsInfoMap();
ColsExtsInfoMap::iterator it = colsExtsInfoMap.begin();
ColExtsInfo::iterator aIt;
std::vector<BRM::FileInfo> files;
BRM::FileInfo aFile;
vector<BRM::LBID_t> lbidList;
BRM::LBID_t startLbid;
while (it != colsExtsInfoMap.end())
{
aIt = (it->second).begin();
aFile.oid = it->first;
while (aIt != (it->second).end())
{
aFile.partitionNum = aIt->partNum;
aFile.dbRoot = aIt->dbRoot;
aFile.segmentNum = aIt->segNum;
aFile.compType = aIt->compType;
files.push_back(aFile);
fDbrm.lookupLocalStartLbid(aFile.oid, aFile.partitionNum, aFile.segmentNum, aIt->hwm, startLbid);
// cout <<"Added to files oid:dbroot:part:seg:compType = " <<
// aFile.oid<<":"<<aFile.dbRoot<<":"<<aFile.partitionNum<<":"<<aFile.segmentNum
//<<":"<<aFile.compType <<endl;
aIt++;
}
it++;
}
cacheutils::purgePrimProcFdCache(files, Config::getLocalModuleID());
fDbrm.invalidateUncommittedExtentLBIDs(0, false, &lbidList);
}
TableMetaData::removeTableMetaData(SYSCOLUMN_BASE);
}
} // namespace WriteEngine
View Git Blame Copy Permalink