database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-04-18 21:44:02 +03:00
Code
mariadb-columnstore-engine/writeengine/wrapper/writeengine.cpp
Daniel Black c01ab42e87 MCOL-5747 gcc-14.1.1 compile error - calloc - transposed args 
The arguments of calloc are the number of members and the sizeof the
member. Gcc-14.1.1 worked out how to tell the difference.

We correct this by transposing to gcc's will.
2024-05-16 16:53:52 +04:00

6736 lines
218 KiB
C++
Raw Permalink Blame History

/* Copyright (C) 2014 InfiniDB, Inc.
Copyright (C) 2016-2021 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: writeengine.cpp 4737 2013-08-14 20:45:46Z bwilkinson $
/** @writeengine.cpp
* A wrapper class for the write engine to write information to files
*/
// XXX: a definition to switch off computations for token columns.
#define XXX_WRITEENGINE_TOKENS_RANGES_XXX
#include <cmath>
#include <cstdlib>
#include <unistd.h>
#include <boost/scoped_array.hpp>
#include <boost/scoped_ptr.hpp>
#include <unordered_map>
using namespace std;
#include "joblisttypes.h"
#define WRITEENGINEWRAPPER_DLLEXPORT
#include "writeengine.h"
#undef WRITEENGINEWRAPPER_DLLEXPORT
#include "we_convertor.h"
#include "we_log.h"
#include "we_simplesyslog.h"
#include "we_config.h"
#include "we_bulkrollbackmgr.h"
#include "brm.h"
#include "stopwatch.h"
#include "we_colop.h"
#include "we_type.h"
#include "we_colopcompress.h"
#include "we_dctnrycompress.h"
#include "cacheutils.h"
#include "calpontsystemcatalog.h"
#include "we_simplesyslog.h"
using namespace cacheutils;
using namespace logging;
using namespace BRM;
using namespace execplan;
#include "IDBDataFile.h"
#include "IDBPolicy.h"
#include "MonitorProcMem.h"
using namespace idbdatafile;
#include "dataconvert.h"
#include "string_prefixes.h"
#include "mcs_decimal.h"
namespace WriteEngine
//#define PROFILE 1
#define RETURN_ON_ERROR_REPORT(statement) \
do \
{ \
int rc = (statement); \
if (rc != NO_ERROR) \
{ \
cerr << "failed at " << __LINE__ << "function " << __FUNCTION__ << ", statement '" << #statement \
<< "', rc " << rc << endl; \
return rc; \
} \
} while (0)
{
StopWatch timer;
using OidToIdxMap = std::unordered_map<OID, ColStructList::size_type>;
/**@brief WriteEngineWrapper Constructor
*/
WriteEngineWrapper::WriteEngineWrapper() : m_opType(NOOP)
{
m_colOp[UN_COMPRESSED_OP] = new ColumnOpCompress0;
m_dctnry[UN_COMPRESSED_OP] = new DctnryCompress0;
m_colOp[COMPRESSED_OP_1] = new ColumnOpCompress1(/*comressionType=*/1);
m_dctnry[COMPRESSED_OP_1] = new DctnryCompress1(/*compressionType=*/1);
m_colOp[COMPRESSED_OP_2] = new ColumnOpCompress1(/*comressionType=*/3);
m_dctnry[COMPRESSED_OP_2] = new DctnryCompress1(/*compressionType=*/3);
}
WriteEngineWrapper::WriteEngineWrapper(const WriteEngineWrapper& rhs) : m_opType(rhs.m_opType)
{
m_colOp[UN_COMPRESSED_OP] = new ColumnOpCompress0;
m_dctnry[UN_COMPRESSED_OP] = new DctnryCompress0;
m_colOp[COMPRESSED_OP_1] = new ColumnOpCompress1(/*compressionType=*/1);
m_dctnry[COMPRESSED_OP_1] = new DctnryCompress1(/*compressionType=*/1);
m_colOp[COMPRESSED_OP_2] = new ColumnOpCompress1(/*compressionType=*/3);
m_dctnry[COMPRESSED_OP_2] = new DctnryCompress1(/*compressionType=*/3);
}
/**@brief WriteEngineWrapper Constructor
*/
WriteEngineWrapper::~WriteEngineWrapper()
{
delete m_colOp[UN_COMPRESSED_OP];
delete m_dctnry[UN_COMPRESSED_OP];
delete m_colOp[COMPRESSED_OP_1];
delete m_dctnry[COMPRESSED_OP_1];
delete m_colOp[COMPRESSED_OP_2];
delete m_dctnry[COMPRESSED_OP_2];
}
/**@brief Perform upfront initialization
*/
/* static */ void WriteEngineWrapper::init(unsigned subSystemID)
{
SimpleSysLog::instance()->setLoggingID(logging::LoggingID(subSystemID));
Config::initConfigCache();
BRMWrapper::getInstance();
// Bug 5415 Add HDFS MemBuffer vs. FileBuffer decision logic.
config::Config* cf = config::Config::makeConfig();
//--------------------------------------------------------------------------
// Memory overload protection. This setting will cause the process to die should
// it, by itself, consume maxPct of total memory. Monitored in MonitorProcMem.
// Only used at the express direction of Field Support.
//--------------------------------------------------------------------------
int maxPct = 0; // disable by default
string strMaxPct = cf->getConfig("WriteEngine", "MaxPct");
if (strMaxPct.length() != 0)
maxPct = cf->uFromText(strMaxPct);
//--------------------------------------------------------------------------
// MemoryCheckPercent. This controls at what percent of total memory be consumed
// by all processes before we switch from HdfsRdwrMemBuffer to HdfsRdwrFileBuffer.
// This is only used in Hdfs installations.
//--------------------------------------------------------------------------
int checkPct = 95;
string strCheckPct = cf->getConfig("SystemConfig", "MemoryCheckPercent");
if (strCheckPct.length() != 0)
checkPct = cf->uFromText(strCheckPct);
//--------------------------------------------------------------------------
// If we're either HDFS, or maxPct is turned on, start the monitor thread.
// Otherwise, we don't need it, so don't waste the resources.
//--------------------------------------------------------------------------
if (maxPct > 0 || IDBPolicy::useHdfs())
{
new boost::thread(utils::MonitorProcMem(maxPct, checkPct, subSystemID));
}
}
/*@brief checkValid --Check input parameters are valid
*/
/***********************************************************
* DESCRIPTION:
* Check input parameters are valid
* PARAMETERS:
* colStructList - column struct list
* colValueList - column value list
* ridList - rowid list
* RETURN:
* NO_ERROR if success
* others if something wrong in the checking process
***********************************************************/
int WriteEngineWrapper::checkValid(const TxnID& txnid, const ColStructList& colStructList,
const ColValueList& colValueList, const RIDList& ridList) const
{
ColTupleList curTupleList;
ColStructList::size_type structListSize;
ColValueList::size_type valListSize;
ColTupleList::size_type totalRow;
if (colStructList.size() == 0)
return ERR_STRUCT_EMPTY;
structListSize = colStructList.size();
valListSize = colValueList.size();
if (structListSize != valListSize)
return ERR_STRUCT_VALUE_NOT_MATCH;
for (ColValueList::size_type i = 0; i < valListSize; i++)
{
curTupleList = static_cast<ColTupleList>(colValueList[i]);
totalRow = curTupleList.size();
if (ridList.size() > 0)
{
if (totalRow != ridList.size())
return ERR_ROWID_VALUE_NOT_MATCH;
}
} // end of for (int i = 0;
return NO_ERROR;
}
/*@brief findSmallestColumn --Find the smallest column for this table
*/
/***********************************************************
* DESCRIPTION:
* Find the smallest column for this table
* PARAMETERS:
* lowColLen - returns smallest column width
* colId - returns smallest column id
* colStructList - column struct list
* RETURN:
* void
***********************************************************/
void WriteEngineWrapper::findSmallestColumn(uint32_t& colId, ColStructList colStructList)
// MCOL-1675: find the smallest column width to calculate the RowID from so
// that all HWMs will be incremented by this operation
{
int32_t lowColLen = 8192;
for (uint32_t colIt = 0; colIt < colStructList.size(); colIt++)
{
if (colStructList[colIt].colWidth < lowColLen)
{
colId = colIt;
lowColLen = colStructList[colId].colWidth;
if (lowColLen == 1)
{
break;
}
}
}
}
/** @brief Fetch values from arrays into VT-types references casting arrays to (element type) ET-typed arrays.
*
* There might be two arrays: one from which we write to buffer and one with old values written before.
* One of these arrays can be absent because of, well, physical absence. Insertion does not have
* values written before and deletion does not have values to write to.
*/
template <typename VT, typename ET>
void fetchNewOldValues(VT& value, VT& oldValue, const void* array, const void* oldArray, size_t i,
size_t totalNewRow)
{
const ET* eArray = static_cast<const ET*>(array);
const ET* oldEArray = static_cast<const ET*>(oldArray);
if (eArray)
{
value = eArray[i < totalNewRow ? i : 0];
}
if (oldEArray)
{
oldValue = oldEArray[i];
}
}
static bool updateBigRangeCheckForInvalidity(ExtCPInfo* maxMin, int128_t value, int128_t oldValue,
const void* valArrayVoid, const void* oldValArrayVoid)
{
if (!oldValArrayVoid)
{ // insertion. we can update range directly. range will not become invalid here.
maxMin->fCPInfo.bigMax =
std::max(maxMin->fCPInfo.bigMax,
value); // we update big range because int columns can be associated with decimals.
maxMin->fCPInfo.bigMin = std::min(maxMin->fCPInfo.bigMin, value);
}
else if (!valArrayVoid)
{ // deletion. we need to check old value only, is it on (or outside) range boundary.
if (oldValue >= maxMin->fCPInfo.bigMax || oldValue <= maxMin->fCPInfo.bigMin)
{
maxMin->toInvalid();
return true; // no point working further.
}
}
else if (valArrayVoid && oldValArrayVoid)
{ // update. we need to check boundaries as in deletion and extend as in insertion.
// check boundaries as in deletion accounting for possible extension.
if ((oldValue <= maxMin->fCPInfo.bigMin && value > oldValue) ||
(oldValue >= maxMin->fCPInfo.bigMax && value < oldValue))
{ // we are overwriting value on the boundary with value that does not preserve or extend range.
maxMin->toInvalid();
return true;
}
maxMin->fCPInfo.bigMax = std::max(maxMin->fCPInfo.bigMax, value);
maxMin->fCPInfo.bigMin = std::min(maxMin->fCPInfo.bigMin, value);
}
return false;
}
template <typename InternalType>
bool updateRangeCheckForInvalidity(ExtCPInfo* maxMin, InternalType value, InternalType oldValue,
const void* valArrayVoid, const void* oldValArrayVoid)
{
if (!oldValArrayVoid)
{ // insertion. we can update range directly. range will not become invalid here.
maxMin->fCPInfo.bigMax = std::max(
maxMin->fCPInfo.bigMax,
(int128_t)value); // we update big range because int columns can be associated with decimals.
maxMin->fCPInfo.bigMin = std::min(maxMin->fCPInfo.bigMin, (int128_t)value);
maxMin->fCPInfo.max = std::max((InternalType)maxMin->fCPInfo.max, value);
maxMin->fCPInfo.min = std::min((InternalType)maxMin->fCPInfo.min, value);
}
else if (!valArrayVoid)
{ // deletion. we need to check old value only, is it on (or outside) range boundary.
if (oldValue >= (InternalType)maxMin->fCPInfo.max || oldValue <= (InternalType)maxMin->fCPInfo.min)
{
maxMin->toInvalid();
return true; // no point working further.
}
}
else if (valArrayVoid && oldValArrayVoid)
{ // update. we need to check boundaries as in deletion and extend as in insertion.
// check boundaries as in deletion accounting for possible extension.
if ((oldValue <= (InternalType)maxMin->fCPInfo.min && value > oldValue) ||
(oldValue >= (InternalType)maxMin->fCPInfo.max && value < oldValue))
{ // we are overwriting value on the boundary with value that does not preserve or extend range.
maxMin->toInvalid();
return true;
}
maxMin->fCPInfo.bigMax = std::max(maxMin->fCPInfo.bigMax, (int128_t)value);
maxMin->fCPInfo.bigMin = std::min(maxMin->fCPInfo.bigMin, (int128_t)value);
maxMin->fCPInfo.max = std::max((InternalType)maxMin->fCPInfo.max, value);
maxMin->fCPInfo.min = std::min((InternalType)maxMin->fCPInfo.min, value);
}
return false;
}
/**
* There can be case with missing valArray (delete), missing oldValArray (insert) and when
* both arrays are present (update).
*/
void WriteEngineWrapper::updateMaxMinRange(const size_t totalNewRow, const size_t totalOldRow,
const execplan::CalpontSystemCatalog::ColType& cscColType,
const ColType colType, const void* valArrayVoid,
const void* oldValArrayVoid, ExtCPInfo* maxMin,
bool canStartWithInvalidRange)
{
if (!maxMin)
{
return;
}
if (colType == WR_CHAR)
{
maxMin->toInvalid(); // simple and wrong solution.
return;
}
bool isUnsigned = false; // TODO: should change with type.
switch (colType)
{
case WR_UBYTE:
case WR_USHORT:
case WR_UINT:
case WR_ULONGLONG:
case WR_CHAR:
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
case WR_TOKEN:
#endif
{
isUnsigned = true;
break;
}
default: // WR_BINARY is signed.
break;
}
if (!canStartWithInvalidRange)
{
// check if range is invalid, we can't update it.
if (maxMin->isInvalid())
{
return;
}
}
if (colType == WR_CHAR)
{
idbassert(MAX_COLUMN_BOUNDARY == sizeof(uint64_t)); // have to check that - code below depends on that.
if (!maxMin->isInvalid())
{
maxMin->fromToChars();
}
}
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
if (colType == WR_TOKEN)
{
oldValArrayVoid = nullptr; // no old values for tokens, sadly.
valArrayVoid = (void*)maxMin->stringsPrefixes();
}
#endif
size_t i;
for (i = 0; i < totalOldRow; i++)
{
int64_t value = 0, oldValue = 0;
uint64_t uvalue = 0, oldUValue = 0;
int128_t bvalue = 0, oldBValue = 0;
// fetching. May not assign value or oldValue variables when array is not present.
switch (colType)
{
case WR_BYTE:
{
fetchNewOldValues<int64_t, int8_t>(value, oldValue, valArrayVoid, oldValArrayVoid, i, totalNewRow);
break;
}
case WR_UBYTE:
{
fetchNewOldValues<uint64_t, uint8_t>(uvalue, oldUValue, valArrayVoid, oldValArrayVoid, i,
totalNewRow);
break;
}
case WR_SHORT:
{
fetchNewOldValues<int64_t, int16_t>(value, oldValue, valArrayVoid, oldValArrayVoid, i, totalNewRow);
break;
}
case WR_USHORT:
{
fetchNewOldValues<uint64_t, uint16_t>(uvalue, oldUValue, valArrayVoid, oldValArrayVoid, i,
totalNewRow);
break;
}
case WR_MEDINT:
case WR_INT:
{
fetchNewOldValues<int64_t, int>(value, oldValue, valArrayVoid, oldValArrayVoid, i, totalNewRow);
break;
}
case WR_UMEDINT:
case WR_UINT:
{
fetchNewOldValues<uint64_t, unsigned int>(uvalue, oldUValue, valArrayVoid, oldValArrayVoid, i,
totalNewRow);
break;
}
case WR_LONGLONG:
{
fetchNewOldValues<int64_t, int64_t>(value, oldValue, valArrayVoid, oldValArrayVoid, i, totalNewRow);
break;
}
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
case WR_TOKEN:
#endif
case WR_ULONGLONG:
{
fetchNewOldValues<uint64_t, uint64_t>(uvalue, oldUValue, valArrayVoid, oldValArrayVoid, i,
totalNewRow);
break;
}
case WR_BINARY:
{
fetchNewOldValues<int128_t, int128_t>(bvalue, oldBValue, valArrayVoid, oldValArrayVoid, i,
totalNewRow);
break;
}
case WR_CHAR:
{
fetchNewOldValues<uint64_t, uint64_t>(uvalue, oldUValue, valArrayVoid, oldValArrayVoid, i, totalNewRow);
// for characters (strings, actually), we fetched then in LSB order, on x86, at the very least.
// this means most significant byte of the string, which is first, is now in LSB of uvalue/oldValue.
// we must perform a conversion.
uvalue = uint64ToStr(uvalue);
oldUValue = uint64ToStr(oldUValue);
break;
}
default: idbassert_s(0, "unknown WR type tag"); return;
}
if (maxMin->isBinaryColumn())
{ // special case of wide decimals. They fit into int128_t range so we do not care about signedness.
if (updateBigRangeCheckForInvalidity(maxMin, bvalue, oldBValue, valArrayVoid, oldValArrayVoid))
{
return;
}
}
else if (isUnsigned)
{
if (updateRangeCheckForInvalidity(maxMin, uvalue, oldUValue, valArrayVoid, oldValArrayVoid))
{
return;
}
}
else
{
if (updateRangeCheckForInvalidity(maxMin, value, oldValue, valArrayVoid, oldValArrayVoid))
{
return;
}
}
}
// the range will be kept.
// to handle character columns properly we need to convert range values back to strings.
if (colType == WR_CHAR)
{
maxMin->fromToChars();
}
}
/*@convertValArray - Convert interface values to internal values
*/
/***********************************************************
* DESCRIPTION:
* Convert interface values to internal values
* PARAMETERS:
* cscColType - CSC ColType struct list
* colStructList - column struct list
* colValueList - column value list
* RETURN:
* none
* valArray - output value array
* nullArray - output null flag array
***********************************************************/
void WriteEngineWrapper::convertValArray(const size_t totalRow,
const CalpontSystemCatalog::ColType& cscColType,
const ColType colType, ColTupleList& curTupleList, void* valArray,
bool bFromList)
{
ColTuple curTuple;
ColTupleList::size_type i;
if (bFromList)
{
for (i = 0; i < curTupleList.size(); i++)
{
curTuple = curTupleList[i];
convertValue(cscColType, colType, valArray, i, curTuple.data, true);
}
}
else
{
for (i = 0; i < totalRow; i++)
{
convertValue(cscColType, colType, valArray, i, curTuple.data, false);
curTupleList.push_back(curTuple);
}
}
}
/*
* @brief Convert column value to its internal representation
*/
void WriteEngineWrapper::convertValue(const execplan::CalpontSystemCatalog::ColType& cscColType,
ColType colType, void* value, boost::any& data)
{
string curStr;
int size;
switch (colType)
{
case WriteEngine::WR_INT:
case WriteEngine::WR_MEDINT:
if (data.type() == typeid(int))
{
int val = boost::any_cast<int>(data);
size = sizeof(int);
memcpy(value, &val, size);
}
else
{
uint32_t val = boost::any_cast<uint32_t>(data);
size = sizeof(uint32_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_UINT:
case WriteEngine::WR_UMEDINT:
{
uint32_t val = boost::any_cast<uint32_t>(data);
size = sizeof(uint32_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT:
curStr = boost::any_cast<string>(data);
if ((int)curStr.length() > MAX_COLUMN_BOUNDARY)
curStr = curStr.substr(0, MAX_COLUMN_BOUNDARY);
memcpy(value, curStr.c_str(), curStr.length());
break;
case WriteEngine::WR_FLOAT:
{
float val = boost::any_cast<float>(data);
// N.B.There is a bug in boost::any or in gcc where, if you store a nan, you will get back a nan,
// but not necessarily the same bits that you put in. This only seems to be for float (double seems
// to work).
if (isnan(val))
{
uint32_t ti = joblist::FLOATNULL;
float* tfp = (float*)&ti;
val = *tfp;
}
size = sizeof(float);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_DOUBLE:
{
double val = boost::any_cast<double>(data);
size = sizeof(double);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_SHORT:
{
short val = boost::any_cast<short>(data);
size = sizeof(short);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_USHORT:
{
uint16_t val = boost::any_cast<uint16_t>(data);
size = sizeof(uint16_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_BYTE:
{
char val = boost::any_cast<char>(data);
size = sizeof(char);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_UBYTE:
{
uint8_t val = boost::any_cast<uint8_t>(data);
size = sizeof(uint8_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_LONGLONG:
if (data.type() == typeid(long long))
{
long long val = boost::any_cast<long long>(data);
size = sizeof(long long);
memcpy(value, &val, size);
}
else if (data.type() == typeid(long))
{
long val = boost::any_cast<long>(data);
size = sizeof(long);
memcpy(value, &val, size);
}
else
{
uint64_t val = boost::any_cast<uint64_t>(data);
size = sizeof(uint64_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_ULONGLONG:
{
uint64_t val = boost::any_cast<uint64_t>(data);
size = sizeof(uint64_t);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_TOKEN:
{
Token val = boost::any_cast<Token>(data);
size = sizeof(Token);
memcpy(value, &val, size);
}
break;
case WriteEngine::WR_BINARY:
{
size = cscColType.colWidth;
int128_t val = boost::any_cast<int128_t>(data);
memcpy(value, &val, size);
}
break;
} // end of switch (colType)
} /*@convertValue - The base for converting values */
/***********************************************************
* DESCRIPTION:
* The base for converting values
* PARAMETERS:
* colType - data type
* pos - array position
* data - value
* RETURN:
* none
***********************************************************/
void WriteEngineWrapper::convertValue(const CalpontSystemCatalog::ColType& cscColType, const ColType colType,
void* valArray, const size_t pos, boost::any& data, bool fromList)
{
string curStr;
if (fromList)
{
switch (colType)
{
case WriteEngine::WR_INT:
case WriteEngine::WR_MEDINT:
// here we assign value to an array element and update range.
if (data.type() == typeid(long))
((int*)valArray)[pos] = static_cast<int>(boost::any_cast<long>(data));
else if (data.type() == typeid(int))
((int*)valArray)[pos] = boost::any_cast<int>(data);
else
{ // this interesting part is for magic values like NULL or EMPTY (marks deleted elements).
// we will not put these into range.
((int*)valArray)[pos] = boost::any_cast<uint32_t>(data);
}
break;
case WriteEngine::WR_UINT:
case WriteEngine::WR_UMEDINT: ((uint32_t*)valArray)[pos] = boost::any_cast<uint32_t>(data); break;
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT:
curStr = boost::any_cast<string>(data);
if ((int)curStr.length() > MAX_COLUMN_BOUNDARY)
curStr = curStr.substr(0, MAX_COLUMN_BOUNDARY);
memcpy((char*)valArray + pos * MAX_COLUMN_BOUNDARY, curStr.c_str(), curStr.length());
break;
// case WriteEngine::WR_LONG : ((long*)valArray)[pos] =
// boost::any_cast<long>(curTuple.data);
// break;
case WriteEngine::WR_FLOAT:
((float*)valArray)[pos] = boost::any_cast<float>(data);
if (isnan(((float*)valArray)[pos]))
{
uint32_t ti = joblist::FLOATNULL;
float* tfp = (float*)&ti;
((float*)valArray)[pos] = *tfp;
}
break;
case WriteEngine::WR_DOUBLE: ((double*)valArray)[pos] = boost::any_cast<double>(data); break;
case WriteEngine::WR_SHORT: ((short*)valArray)[pos] = boost::any_cast<short>(data); break;
case WriteEngine::WR_USHORT:
((uint16_t*)valArray)[pos] = boost::any_cast<uint16_t>(data);
break;
// case WriteEngine::WR_BIT: ((bool*)valArray)[pos] = boost::any_cast<bool>(data);
// break;
case WriteEngine::WR_BYTE: ((char*)valArray)[pos] = boost::any_cast<char>(data); break;
case WriteEngine::WR_UBYTE: ((uint8_t*)valArray)[pos] = boost::any_cast<uint8_t>(data); break;
case WriteEngine::WR_LONGLONG:
if (data.type() == typeid(long long))
((long long*)valArray)[pos] = boost::any_cast<long long>(data);
else if (data.type() == typeid(long))
((long long*)valArray)[pos] = (long long)boost::any_cast<long>(data);
else
((long long*)valArray)[pos] = boost::any_cast<uint64_t>(data);
break;
case WriteEngine::WR_ULONGLONG: ((uint64_t*)valArray)[pos] = boost::any_cast<uint64_t>(data); break;
case WriteEngine::WR_TOKEN:
((Token*)valArray)[pos] = boost::any_cast<Token>(data);
break;
case WriteEngine::WR_BINARY:
size_t size = cscColType.colWidth;
int128_t val = boost::any_cast<int128_t>(data);
memcpy((uint8_t*)valArray + pos * size, &val, size);
break;
} // end of switch (colType)
}
else
{
switch (colType)
{
case WriteEngine::WR_INT:
case WriteEngine::WR_MEDINT: data = ((int*)valArray)[pos]; break;
case WriteEngine::WR_UINT:
case WriteEngine::WR_UMEDINT: data = ((uint64_t*)valArray)[pos]; break;
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT:
char tmp[10];
memcpy(tmp, (char*)valArray + pos * 8, 8);
curStr = tmp;
data = curStr;
break;
// case WriteEngine::WR_LONG : ((long*)valArray)[pos] =
// boost::any_cast<long>(curTuple.data);
// break;
case WriteEngine::WR_FLOAT: data = ((float*)valArray)[pos]; break;
case WriteEngine::WR_DOUBLE: data = ((double*)valArray)[pos]; break;
case WriteEngine::WR_SHORT: data = ((short*)valArray)[pos]; break;
case WriteEngine::WR_USHORT:
data = ((uint16_t*)valArray)[pos];
break;
// case WriteEngine::WR_BIT: data = ((bool*)valArray)[pos];
// break;
case WriteEngine::WR_BYTE: data = ((char*)valArray)[pos]; break;
case WriteEngine::WR_UBYTE: data = ((uint8_t*)valArray)[pos]; break;
case WriteEngine::WR_LONGLONG: data = ((long long*)valArray)[pos]; break;
case WriteEngine::WR_ULONGLONG: data = ((uint64_t*)valArray)[pos]; break;
case WriteEngine::WR_TOKEN: data = ((Token*)valArray)[pos]; break;
case WriteEngine::WR_BINARY: data = ((int128_t*)valArray)[pos]; break;
} // end of switch (colType)
} // end of if
}
/*@createColumn - Create column files, including data and bitmap files
*/
/***********************************************************
* DESCRIPTION:
* Create column files, including data and bitmap files
* PARAMETERS:
* dataOid - column data file id
* bitmapOid - column bitmap file id
* colWidth - column width
* dbRoot - DBRoot where file is to be located
* partition - Starting partition number for segment file path
* compressionType - compression type
* RETURN:
* NO_ERROR if success
* ERR_FILE_EXIST if file exists
* ERR_FILE_CREATE if something wrong in creating the file
***********************************************************/
int WriteEngineWrapper::createColumn(const TxnID& txnid, const OID& dataOid,
const CalpontSystemCatalog::ColDataType dataType, int dataWidth,
uint16_t dbRoot, uint32_t partition, int compressionType)
{
int rc;
Column curCol;
int compress_op = op(compressionType);
m_colOp[compress_op]->initColumn(curCol);
m_colOp[compress_op]->findTypeHandler(dataWidth, dataType);
rc = m_colOp[compress_op]->createColumn(curCol, 0, dataWidth, dataType, WriteEngine::WR_CHAR, (FID)dataOid,
dbRoot, partition);
// This is optional, however, it's recommended to do so to free heap
// memory if assigned in the future
m_colOp[compress_op]->clearColumn(curCol);
std::map<FID, FID> oids;
if (rc == NO_ERROR)
rc = flushDataFiles(NO_ERROR, txnid, oids);
if (rc != NO_ERROR)
{
return rc;
}
RETURN_ON_ERROR(BRMWrapper::getInstance()->setLocalHWM(dataOid, partition, 0, 0));
// @bug 281 : fix for bug 281 - Add flush VM cache to clear all write buffer
// flushVMCache();
return rc;
}
// BUG931
/**
* @brief Fill column with default values
*/
int WriteEngineWrapper::fillColumn(const TxnID& txnid, const OID& dataOid,
const CalpontSystemCatalog::ColType& colType, ColTuple defaultVal,
const OID& refColOID,
const CalpontSystemCatalog::ColDataType refColDataType, int refColWidth,
int refCompressionType, bool isNULL, int compressionType,
const string& defaultValStr, const OID& dictOid, bool autoincrement)
{
int rc = NO_ERROR;
Column newCol;
Column refCol;
ColType newColType;
ColType refColType;
boost::scoped_array<char> defVal(new char[datatypes::MAXDECIMALWIDTH]);
ColumnOp* colOpNewCol = m_colOp[op(compressionType)];
ColumnOp* refColOp = m_colOp[op(refCompressionType)];
Dctnry* dctnry = m_dctnry[op(compressionType)];
colOpNewCol->initColumn(newCol);
refColOp->initColumn(refCol);
uint16_t dbRoot = 1; // not to be used
int newDataWidth = colType.colWidth;
// Convert HWM of the reference column for the new column
// Bug 1703,1705
bool isToken = false;
if (((colType.colDataType == CalpontSystemCatalog::VARCHAR) && (colType.colWidth > 7)) ||
((colType.colDataType == CalpontSystemCatalog::CHAR) && (colType.colWidth > 8)) ||
(colType.colDataType == CalpontSystemCatalog::VARBINARY) ||
(colType.colDataType == CalpontSystemCatalog::BLOB) ||
(colType.colDataType == CalpontSystemCatalog::TEXT))
{
isToken = true;
}
Convertor::convertColType(colType.colDataType, colType.colWidth, newColType, isToken);
// WIP
// replace with isDictCol
if (((refColDataType == CalpontSystemCatalog::VARCHAR) && (refColWidth > 7)) ||
((refColDataType == CalpontSystemCatalog::CHAR) && (refColWidth > 8)) ||
(refColDataType == CalpontSystemCatalog::VARBINARY) ||
(colType.colDataType == CalpontSystemCatalog::BLOB) ||
(colType.colDataType == CalpontSystemCatalog::TEXT))
{
isToken = true;
}
newDataWidth = colOpNewCol->getCorrectRowWidth(colType.colDataType, colType.colWidth);
// MCOL-1347 CS doubles the width for ALTER TABLE..ADD COLUMN
if (colType.colWidth < 4 && colType.colDataType == CalpontSystemCatalog::VARCHAR)
{
newDataWidth >>= 1;
}
Convertor::convertColType(refColDataType, refColWidth, refColType, isToken);
refColOp->setColParam(refCol, 0, refColOp->getCorrectRowWidth(refColDataType, refColWidth), refColDataType,
refColType, (FID)refColOID, refCompressionType, dbRoot);
colOpNewCol->setColParam(newCol, 0, newDataWidth, colType.colDataType, newColType, (FID)dataOid,
compressionType, dbRoot);
// refColOp and colOpNewCol are the same ptr.
colOpNewCol->findTypeHandler(newDataWidth, colType.colDataType);
int size = sizeof(Token);
if (newColType == WriteEngine::WR_TOKEN)
{
if (isNULL)
{
Token nullToken;
memcpy(defVal.get(), &nullToken, size);
}
// Tokenization is done when we create dictionary file
}
else
{
// WIP
convertValue(colType, newColType, defVal.get(), defaultVal.data);
}
if (rc == NO_ERROR)
rc = colOpNewCol->fillColumn(txnid, newCol, refCol, defVal.get(), dctnry, refColOp, dictOid,
colType.colWidth, defaultValStr, autoincrement);
// flushing files is in colOp->fillColumn()
return rc;
}
// TODO: Get rid of this
void emptyValueToAny(boost::any* any, const uint8_t* emptyValue, int colWidth)
{
switch (colWidth)
{
case 16:
*any = *(uint128_t*)emptyValue;
break;
case 8:
*any = *(uint64_t*)emptyValue;
break;
case 4:
*any = *(uint32_t*)emptyValue;
break;
case 2:
*any = *(uint16_t*)emptyValue;
break;
default:
*any = *emptyValue;
}
}
int WriteEngineWrapper::deleteRow(const TxnID& txnid, const vector<CSCTypesList>& colExtentsColType,
vector<ColStructList>& colExtentsStruct, vector<void*>& colOldValueList,
vector<RIDList>& ridLists, const int32_t tableOid, bool hasAUXCol)
{
ColTuple curTuple;
ColStruct curColStruct;
CalpontSystemCatalog::ColType cscColType;
DctnryStruct dctnryStruct;
ColValueList colValueList;
ColTupleList curTupleList;
DctnryStructList dctnryStructList;
DctnryValueList dctnryValueList;
ColStructList colStructList;
CSCTypesList cscColTypeList;
int rc;
string tmpStr("");
vector<DctnryStructList> dctnryExtentsStruct;
if (colExtentsStruct.size() == 0 || ridLists.size() == 0)
return ERR_STRUCT_EMPTY;
// set transaction id
setTransId(txnid);
unsigned numExtents = colExtentsStruct.size();
for (unsigned extent = 0; extent < numExtents; extent++)
{
colStructList = colExtentsStruct[extent];
cscColTypeList = colExtentsColType[extent];
for (ColStructList::size_type i = 0; i < colStructList.size(); i++)
{
curTupleList.clear();
curColStruct = colStructList[i];
cscColType = cscColTypeList[i];
Convertor::convertColType(&curColStruct);
const uint8_t* emptyVal = m_colOp[op(curColStruct.fCompressionType)]->getEmptyRowValue(
curColStruct.colDataType, curColStruct.colWidth);
emptyValueToAny(&curTuple.data, emptyVal, curColStruct.colWidth);
curTupleList.push_back(curTuple);
colValueList.push_back(curTupleList);
dctnryStruct.dctnryOid = 0;
dctnryStruct.fColPartition = curColStruct.fColPartition;
dctnryStruct.fColSegment = curColStruct.fColSegment;
dctnryStruct.fColDbRoot = curColStruct.fColDbRoot;
dctnryStruct.columnOid = colStructList[i].dataOid;
dctnryStructList.push_back(dctnryStruct);
DctnryTuple dctnryTuple;
DctColTupleList dctColTuples;
dctnryTuple.sigValue = (unsigned char*)tmpStr.c_str();
dctnryTuple.sigSize = tmpStr.length();
dctnryTuple.isNull = true;
dctColTuples.push_back(dctnryTuple);
dctnryValueList.push_back(dctColTuples);
}
dctnryExtentsStruct.push_back(dctnryStructList);
}
// unfortunately I don't have a better way to instruct without passing too many parameters
m_opType = DELETE;
rc = updateColumnRec(txnid, colExtentsColType, colExtentsStruct, colValueList, colOldValueList, ridLists,
dctnryExtentsStruct, dctnryValueList, tableOid, hasAUXCol);
m_opType = NOOP;
return rc;
}
inline void allocateValArray(void*& valArray, ColTupleList::size_type totalRow, ColType colType, int colWidth)
{
// MCS allocates 8 bytes even for CHARs smaller then 8 bytes.
switch (colType)
{
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT: valArray = calloc(totalRow * MAX_COLUMN_BOUNDARY, sizeof(char)); break;
case WriteEngine::WR_TOKEN: valArray = calloc(totalRow, sizeof(Token)); break;
default: valArray = calloc(totalRow, colWidth); break;
}
}
int WriteEngineWrapper::deleteBadRows(const TxnID& txnid, ColStructList& colStructs, RIDList& ridList,
DctnryStructList& dctnryStructList)
{
/* Need to scan all files including dictionary store files to check whether there is any bad chunks
*
*/
int rc = 0;
Column curCol;
void* valArray = NULL;
m_opType = DELETE;
for (unsigned i = 0; i < colStructs.size(); i++)
{
ColumnOp* colOp = m_colOp[op(colStructs[i].fCompressionType)];
unsigned needFixFiles = colStructs[i].tokenFlag ? 2 : 1;
colOp->initColumn(curCol);
for (unsigned j = 0; j < needFixFiles; j++)
{
if (j == 0)
{
colOp->setColParam(curCol, 0, colStructs[i].colWidth, colStructs[i].colDataType,
colStructs[i].colType, colStructs[i].dataOid, colStructs[i].fCompressionType,
colStructs[i].fColDbRoot, colStructs[i].fColPartition, colStructs[i].fColSegment);
colOp->findTypeHandler(colStructs[i].colWidth, colStructs[i].colDataType);
string segFile;
rc = colOp->openColumnFile(curCol, segFile, true, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR) // If openFile fails, disk error or header error is assumed.
{
// report error and return.
std::ostringstream oss;
WErrorCodes ec;
string err = ec.errorString(rc);
oss << "Error opening file oid:dbroot:partition:segment = " << colStructs[i].dataOid << ":"
<< colStructs[i].fColDbRoot << ":" << colStructs[i].fColPartition << ":"
<< colStructs[i].fColSegment << " and error code is " << rc << " with message " << err;
throw std::runtime_error(oss.str());
}
allocateValArray(valArray, 1, colStructs[i].colType, colStructs[i].colWidth);
rc = colOp->writeRows(curCol, ridList.size(), ridList, valArray, 0, true);
if (rc != NO_ERROR)
{
// read error is fixed in place
if (rc == ERR_COMP_COMPRESS) // write error
{
}
}
// flush files will be done in the end of fix.
colOp->clearColumn(curCol);
if (valArray != NULL)
free(valArray);
}
else // dictionary file. How to fix
{
// read headers out, uncompress the last chunk, if error, replace it with empty chunk.
Dctnry* dctnry = m_dctnry[op(dctnryStructList[i].fCompressionType)];
rc = dctnry->openDctnry(dctnryStructList[i].dctnryOid, dctnryStructList[i].fColDbRoot,
dctnryStructList[i].fColPartition, dctnryStructList[i].fColSegment, false);
rc = dctnry->checkFixLastDictChunk();
rc = dctnry->closeDctnry(true);
}
}
}
return rc;
}
/*@flushVMCache - Flush VM cache
*/
/***********************************************************
* DESCRIPTION:
* Flush sytem VM cache
* PARAMETERS:
* none
* RETURN:
* none
***********************************************************/
void WriteEngineWrapper::flushVMCache() const
{
// int fd = open("/proc/sys/vm/drop_caches", O_WRONLY);
// write(fd, "3", 1);
// close(fd);
}
#if 0
// XXX temporary for debugging purposes
static void log_this(const char *message,
logging::LOG_TYPE log_type, unsigned sid)
{
// corresponds with dbcon in SubsystemID vector
// in messagelog.cpp
unsigned int subSystemId = 24;
logging::LoggingID logid( subSystemId, sid, 0);
logging::Message::Args args1;
logging::Message msg(1);
args1.add(message);
msg.format( args1 );
Logger logger(logid.fSubsysID);
logger.logMessage(log_type, msg, logid);
}
#endif
/** @brief Determine whether we may update a column's ranges (by type) and return nullptr if we can't */
static ExtCPInfo* getCPInfoToUpdateForUpdatableType(const ColStruct& colStruct, ExtCPInfo* currentCPInfo,
OpType optype)
{
if (colStruct.tokenFlag)
{
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
if (currentCPInfo && currentCPInfo->hasStringsPrefixes() && optype == INSERT)
{
return currentCPInfo;
}
#endif
return nullptr;
}
switch (colStruct.colType)
{
// here we enumerate all supported types.
case WR_BYTE:
case WR_SHORT:
case WR_INT:
case WR_LONGLONG:
case WR_CHAR:
case WR_UBYTE:
case WR_USHORT:
case WR_UINT:
case WR_ULONGLONG:
case WR_MEDINT:
case WR_UMEDINT:
case WR_BINARY:
{
return currentCPInfo;
}
// all unsupported types must not be supported.
default: return nullptr; // safe choice for everything we can't do.
}
}
/** @brief Let only valid ranges to be present.
*
* There can be a case that we have computed invalid range while computing updated ranges.
*
* These invalid ranges should not have CP_VALID in the status code. So, they must not
* come into final call to setCPInfos or something.
*
* To achieve that, we filter these invalid ranges here.
*/
static void setInvalidCPInfosSpecialMarks(std::vector<ExtCPInfo>& cpInfos)
{
size_t i;
for (i = 0; i < cpInfos.size(); i++)
{
if (cpInfos[i].isInvalid())
{
cpInfos[i].fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
}
}
}
/*@insertColumnRecs - Insert value(s) into a column
*/
/***********************************************************
* DESCRIPTION:
* Insert values into columns (batchinsert)
* PARAMETERS:
* colStructList - column struct list
* colValueList - column value list
* RETURN:
* NO_ERROR if success
* others if something wrong in inserting the value
***********************************************************/
int WriteEngineWrapper::insertColumnRecs(
const TxnID& txnid, const CSCTypesList& cscColTypeList, ColStructList& colStructList,
ColValueList& colValueList, DctnryStructList& dctnryStructList, DictStrList& dictStrList,
std::vector<boost::shared_ptr<DBRootExtentTracker> >& dbRootExtentTrackers, RBMetaWriter* fRBMetaWriter,
bool bFirstExtentOnThisPM, bool insertSelect, bool isAutoCommitOn, OID tableOid, bool isFirstBatchPm)
{
int rc;
RID* rowIdArray = NULL;
ColTupleList curTupleList;
Column curCol;
ColStruct curColStruct;
ColValueList colOldValueList;
ColValueList colNewValueList;
ColStructList newColStructList;
DctnryStructList newDctnryStructList;
HWM hwm = 0;
HWM oldHwm = 0;
HWM newHwm = 0;
ColTupleList::size_type totalRow;
ColStructList::size_type totalColumns;
uint64_t rowsLeft = 0;
bool newExtent = false;
RIDList ridList;
ColumnOp* colOp = NULL;
ColSplitMaxMinInfoList maxMins;
// Set tmp file suffix to modify HDFS db file
bool useTmpSuffix = false;
m_opType = INSERT;
if (idbdatafile::IDBPolicy::useHdfs())
{
if (!bFirstExtentOnThisPM)
useTmpSuffix = true;
}
unsigned i = 0;
#ifdef PROFILE
StopWatch timer;
#endif
// debug information for testing
if (isDebug(DEBUG_2))
{
printInputValue(colStructList, colValueList, ridList, dctnryStructList, dictStrList);
}
// end
// Convert data type and column width to write engine specific
for (i = 0; i < colStructList.size(); i++)
Convertor::convertColType(&colStructList[i]);
for (const auto& colStruct : colStructList)
{
ColSplitMaxMinInfo tmp(colStruct.colDataType, colStruct.colWidth);
maxMins.push_back(tmp);
}
uint32_t colId = 0;
// MCOL-1675: find the smallest column width to calculate the RowID from so
// that all HWMs will be incremented by this operation
findSmallestColumn(colId, colStructList);
// rc = checkValid(txnid, colStructList, colValueList, ridList);
// if (rc != NO_ERROR)
// return rc;
setTransId(txnid);
uint16_t dbRoot, segmentNum;
uint32_t partitionNum;
string segFile;
bool newFile;
TableMetaData* tableMetaData = TableMetaData::makeTableMetaData(tableOid);
// populate colStructList with file information
IDBDataFile* pFile = NULL;
std::vector<DBRootExtentInfo> extentInfo;
int currentDBrootIdx = 0;
std::vector<BRM::CreateStripeColumnExtentsArgOut>
extents; // this structure valid only when isFirstOnBatchPm is true.
std::vector<BRM::LBID_t> newExtentsStartingLbids; // we keep column-indexed LBIDs here for **new** extents.
// It is set and valid and used only when rowsLeft > 0.
//--------------------------------------------------------------------------
// For first batch on this PM:
// o get starting extent from ExtentTracker, and allocate extent if needed
// o construct colStructList and dctnryStructList accordingly
// o save extent information in tableMetaData for future use
// If not first batch on this PM:
// o construct colStructList and dctnryStructList from tableMetaData
//--------------------------------------------------------------------------
if (isFirstBatchPm)
{
currentDBrootIdx = dbRootExtentTrackers[colId]->getCurrentDBRootIdx();
extentInfo = dbRootExtentTrackers[colId]->getDBRootExtentList();
dbRoot = extentInfo[currentDBrootIdx].fDbRoot;
partitionNum = extentInfo[currentDBrootIdx].fPartition;
//----------------------------------------------------------------------
// check whether this extent is the first on this PM
//----------------------------------------------------------------------
if (bFirstExtentOnThisPM)
{
std::vector<BRM::CreateStripeColumnExtentsArgIn> cols;
BRM::CreateStripeColumnExtentsArgIn createStripeColumnExtentsArgIn;
for (i = 0; i < colStructList.size(); i++)
{
createStripeColumnExtentsArgIn.oid = colStructList[i].dataOid;
createStripeColumnExtentsArgIn.width = colStructList[i].colWidth;
createStripeColumnExtentsArgIn.colDataType = colStructList[i].colDataType;
cols.push_back(createStripeColumnExtentsArgIn);
}
rc = BRMWrapper::getInstance()->allocateStripeColExtents(cols, dbRoot, partitionNum, segmentNum,
extents);
if (rc != NO_ERROR)
return rc;
// Create column files
ExtCPInfoList cpinfoList;
for (i = 0; i < extents.size(); i++)
{
ExtCPInfo cpInfo(colStructList[i].colDataType, colStructList[i].colWidth);
colOp = m_colOp[op(colStructList[i].fCompressionType)];
colOp->initColumn(curCol);
colOp->setColParam(curCol, colId, colStructList[i].colWidth, colStructList[i].colDataType,
colStructList[i].colType, colStructList[i].dataOid,
colStructList[i].fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(colStructList[i].colWidth, colStructList[i].colDataType);
rc = colOp->extendColumn(curCol, false, extents[i].startBlkOffset, extents[i].startLbid,
extents[i].allocSize, dbRoot, partitionNum, segmentNum, segFile, pFile,
newFile);
if (rc != NO_ERROR)
return rc;
cpInfo.toInvalid();
cpInfo.fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
// mark the extents to invalid
cpInfo.fCPInfo.firstLbid = extents[i].startLbid;
cpinfoList.push_back(cpInfo);
colStructList[i].fColPartition = partitionNum;
colStructList[i].fColSegment = segmentNum;
colStructList[i].fColDbRoot = dbRoot;
dctnryStructList[i].fColPartition = partitionNum;
dctnryStructList[i].fColSegment = segmentNum;
dctnryStructList[i].fColDbRoot = dbRoot;
}
// mark the extents to invalid
rc = BRMWrapper::getInstance()->setExtentsMaxMin(cpinfoList);
if (rc != NO_ERROR)
return rc;
// create corresponding dictionary files
for (i = 0; i < dctnryStructList.size(); i++)
{
if (dctnryStructList[i].dctnryOid > 0)
{
rc = createDctnry(txnid, dctnryStructList[i].dctnryOid, dctnryStructList[i].colWidth, dbRoot,
partitionNum, segmentNum, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
return rc;
}
}
} // if ( bFirstExtentOnThisPM)
else // if (!bFirstExtentOnThisPM)
{
std::vector<DBRootExtentInfo> tmpExtentInfo;
for (i = 0; i < dbRootExtentTrackers.size(); i++)
{
tmpExtentInfo = dbRootExtentTrackers[i]->getDBRootExtentList();
colStructList[i].fColPartition = tmpExtentInfo[currentDBrootIdx].fPartition;
colStructList[i].fColSegment = tmpExtentInfo[currentDBrootIdx].fSegment;
colStructList[i].fColDbRoot = tmpExtentInfo[currentDBrootIdx].fDbRoot;
dctnryStructList[i].fColPartition = tmpExtentInfo[currentDBrootIdx].fPartition;
dctnryStructList[i].fColSegment = tmpExtentInfo[currentDBrootIdx].fSegment;
dctnryStructList[i].fColDbRoot = tmpExtentInfo[currentDBrootIdx].fDbRoot;
}
}
//----------------------------------------------------------------------
// Save the extents info in tableMetaData
//----------------------------------------------------------------------
for (i = 0; i < colStructList.size(); i++)
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = colStructList[i].fColDbRoot;
aExt.partNum = colStructList[i].fColPartition;
aExt.segNum = colStructList[i].fColSegment;
aExt.compType = colStructList[i].fCompressionType;
aExt.isDict = false;
if (bFirstExtentOnThisPM)
{
aExt.hwm = extents[i].startBlkOffset;
aExt.isNewExt = true;
}
else
{
std::vector<DBRootExtentInfo> tmpExtentInfo;
tmpExtentInfo = dbRootExtentTrackers[i]->getDBRootExtentList();
aExt.isNewExt = false;
aExt.hwm = tmpExtentInfo[currentDBrootIdx].fLocalHwm;
}
aExt.current = true;
aColExtsInfo.push_back(aExt);
}
tableMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
for (i = 0; i < dctnryStructList.size(); i++)
{
if (dctnryStructList[i].dctnryOid > 0)
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(dctnryStructList[i].dctnryOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == dctnryStructList[i].fColDbRoot) &&
(it->partNum == dctnryStructList[i].fColPartition) &&
(it->segNum == dctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = dctnryStructList[i].fColDbRoot;
aExt.partNum = dctnryStructList[i].fColPartition;
aExt.segNum = dctnryStructList[i].fColSegment;
aExt.compType = dctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
}
tableMetaData->setColExtsInfo(dctnryStructList[i].dctnryOid, aColExtsInfo);
}
}
} // if (isFirstBatchPm)
else // get the extent info from tableMetaData
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[colId].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if (it->current)
break;
it++;
}
if (it == aColExtsInfo.end())
return 1;
for (i = 0; i < colStructList.size(); i++)
{
colStructList[i].fColPartition = it->partNum;
colStructList[i].fColSegment = it->segNum;
colStructList[i].fColDbRoot = it->dbRoot;
dctnryStructList[i].fColPartition = it->partNum;
dctnryStructList[i].fColSegment = it->segNum;
dctnryStructList[i].fColDbRoot = it->dbRoot;
}
}
curTupleList = static_cast<ColTupleList>(colValueList[0]);
totalRow = curTupleList.size();
totalColumns = colStructList.size();
rowIdArray = new RID[totalRow];
// use scoped_array to ensure ptr deletion regardless of where we return
boost::scoped_array<RID> rowIdArrayPtr(rowIdArray);
memset(rowIdArray, 0, (sizeof(RID) * totalRow));
//--------------------------------------------------------------------------
// allocate row id(s)
//--------------------------------------------------------------------------
curColStruct = colStructList[colId];
colOp = m_colOp[op(curColStruct.fCompressionType)];
colOp->initColumn(curCol);
// Get the correct segment, partition, column file
vector<ExtentInfo> colExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> dictExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> fileInfo;
dbRoot = curColStruct.fColDbRoot;
// use the first column to calculate row id
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[colId].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[colId].fColDbRoot) &&
(it->partNum == colStructList[colId].fColPartition) &&
(it->segNum == colStructList[colId].fColSegment) && it->current)
{
break;
}
it++;
}
if (it != aColExtsInfo.end())
{
hwm = it->hwm;
}
oldHwm = hwm; // Save this info for rollback
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, colId, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, curColStruct.fColDbRoot,
curColStruct.fColPartition, curColStruct.fColSegment);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
return rc;
}
// get hwm first
// @bug 286 : fix for bug 286 - correct the typo in getHWM
// RETURN_ON_ERROR(BRMWrapper::getInstance()->getHWM(curColStruct.dataOid, hwm));
Column newCol;
#ifdef PROFILE
timer.start("allocRowId");
#endif
newColStructList = colStructList;
newDctnryStructList = dctnryStructList;
bool bUseStartExtent = true;
if (idbdatafile::IDBPolicy::useHdfs())
insertSelect = true;
rc = colOp->allocRowId(txnid, bUseStartExtent, curCol, (uint64_t)totalRow, rowIdArray, hwm, newExtent,
rowsLeft, newHwm, newFile, newColStructList, newDctnryStructList,
dbRootExtentTrackers, insertSelect, true, tableOid, isFirstBatchPm,
&newExtentsStartingLbids);
if (rc != NO_ERROR) // Clean up is already done
return rc;
#ifdef PROFILE
timer.stop("allocRowId");
#endif
//--------------------------------------------------------------------------
// Expand initial abbreviated extent if any RID in 1st extent is > 256K.
// if totalRow == rowsLeft, then not adding rows to 1st extent, so skip it.
//--------------------------------------------------------------------------
// DMC-SHARED_NOTHING_NOTE: Is it safe to assume only part0 seg0 is abbreviated?
if ((curCol.dataFile.fPartition == 0) && (curCol.dataFile.fSegment == 0) && ((totalRow - rowsLeft) > 0) &&
(rowIdArray[totalRow - rowsLeft - 1] >= (RID)INITIAL_EXTENT_ROWS_TO_DISK))
{
for (unsigned k = 0; k < colStructList.size(); k++)
{
// Skip the selected column
if (k == colId)
continue;
Column expandCol;
colOp = m_colOp[op(colStructList[k].fCompressionType)];
colOp->setColParam(expandCol, 0, colStructList[k].colWidth, colStructList[k].colDataType,
colStructList[k].colType, colStructList[k].dataOid,
colStructList[k].fCompressionType, colStructList[k].fColDbRoot,
colStructList[k].fColPartition, colStructList[k].fColSegment);
colOp->findTypeHandler(colStructList[k].colWidth, colStructList[k].colDataType);
rc = colOp->openColumnFile(expandCol, segFile, true); // @bug 5572 HDFS tmp file
if (rc == NO_ERROR)
{
if (colOp->abbreviatedExtent(expandCol.dataFile.pFile, colStructList[k].colWidth))
{
rc = colOp->expandAbbrevExtent(expandCol);
}
}
if (rc != NO_ERROR)
{
return rc;
}
colOp->clearColumn(expandCol); // closes the file (if uncompressed)
}
}
//--------------------------------------------------------------------------
// Tokenize data if needed
//--------------------------------------------------------------------------
if (insertSelect && isAutoCommitOn)
BRMWrapper::setUseVb(false);
else
BRMWrapper::setUseVb(true);
dictStr::iterator dctStr_iter;
ColTupleList::iterator col_iter;
for (i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
dctStr_iter = dictStrList[i].begin();
col_iter = colValueList[i].begin();
Dctnry* dctnry = m_dctnry[op(dctnryStructList[i].fCompressionType)];
rc = dctnry->openDctnry(dctnryStructList[i].dctnryOid, dctnryStructList[i].fColDbRoot,
dctnryStructList[i].fColPartition, dctnryStructList[i].fColSegment,
useTmpSuffix); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
// cout << "Error opening dctnry file " << dctnryStructList[i].dctnryOid << endl;
return rc;
}
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
datatypes::Charset cset(dctnryStructList[i].fCharsetNumber);
#endif
for (uint32_t rows = 0; rows < (totalRow - rowsLeft); rows++)
{
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
int64_t strPrefix;
#endif
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
strPrefix = (int64_t)joblist::UBIGINTNULL; // the string prefixes are signed long ints.
#endif
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
strPrefix = encodeStringPrefix(dctTuple.sigValue, dctTuple.sigSize, cset);
#endif
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
maxMins[i].fSplitMaxMinInfo[0].addStringPrefix(strPrefix);
#endif
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry(false);
if (rc != NO_ERROR)
return rc;
if (newExtent)
{
//@Bug 4854 back up hwm chunk for the file to be modified
if (fRBMetaWriter)
fRBMetaWriter->backupDctnryHWMChunk(
newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment);
rc = dctnry->openDctnry(newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment,
false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
for (uint32_t rows = 0; rows < rowsLeft; rows++)
{
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
int64_t strPrefix;
#endif
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
strPrefix = joblist::UBIGINTNULL; // string prefixes are signed long ints.
#endif
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
strPrefix = encodeStringPrefix_check_null(dctTuple.sigValue, dctTuple.sigSize, cset);
#endif
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, newDctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
#if defined(XXX_WRITEENGINE_TOKENS_RANGES_XXX)
maxMins[i].fSplitMaxMinInfo[1].addStringPrefix(strPrefix);
#endif
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry(false);
if (rc != NO_ERROR)
return rc;
}
}
}
if (insertSelect && isAutoCommitOn)
BRMWrapper::setUseVb(false);
else
BRMWrapper::setUseVb(true);
//--------------------------------------------------------------------------
// Update column info structure @Bug 1862 set hwm, and
// Prepare ValueList for new extent (if applicable)
//--------------------------------------------------------------------------
//@Bug 2205 Check whether all rows go to the new extent
RID lastRid = 0;
RID lastRidNew = 0;
if (totalRow - rowsLeft > 0)
{
lastRid = rowIdArray[totalRow - rowsLeft - 1];
lastRidNew = rowIdArray[totalRow - 1];
}
else
{
lastRid = 0;
lastRidNew = rowIdArray[totalRow - 1];
}
// if a new extent is created, all the columns in this table should have their own new extent
// First column already processed
//@Bug 1701. Close the file (if uncompressed)
m_colOp[op(curCol.compressionType)]->clearColumn(curCol);
// Update hwm to set them in the end
bool succFlag = false;
unsigned colWidth = 0;
int curFbo = 0, curBio;
for (i = 0; i < totalColumns; i++)
{
// shoud be obtained from saved hwm
aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[i].dataOid);
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment) && it->current)
break;
it++;
}
if (it != aColExtsInfo.end()) // update hwm info
{
oldHwm = it->hwm;
}
// save hwm for the old extent
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
if ((HWM)curFbo >= oldHwm)
{
it->hwm = (HWM)curFbo;
}
//@Bug 4947. set current to false for old extent.
if (newExtent)
{
it->current = false;
}
}
else
return ERR_INVALID_PARAM;
// update hwm for the new extent
if (newExtent)
{
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newColStructList[i].fColDbRoot) &&
(it->partNum == newColStructList[i].fColPartition) &&
(it->segNum == newColStructList[i].fColSegment) && it->current)
break;
it++;
}
succFlag = colOp->calculateRowId(lastRidNew, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
if (it != aColExtsInfo.end())
{
it->hwm = (HWM)curFbo;
// cout << "setting hwm to " << (int)curFbo <<" for seg " <<it->segNum << endl;
it->current = true;
}
}
else
return ERR_INVALID_PARAM;
}
tableMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
//--------------------------------------------------------------------------
// Prepare the valuelist for the new extent
//--------------------------------------------------------------------------
ColTupleList colTupleList;
ColTupleList newColTupleList;
ColTupleList firstPartTupleList;
for (unsigned i = 0; i < totalColumns; i++)
{
colTupleList = static_cast<ColTupleList>(colValueList[i]);
for (uint64_t j = rowsLeft; j > 0; j--)
{
newColTupleList.push_back(colTupleList[totalRow - j]);
}
colNewValueList.push_back(newColTupleList);
newColTupleList.clear();
// update the oldvalue list for the old extent
for (uint64_t j = 0; j < (totalRow - rowsLeft); j++)
{
firstPartTupleList.push_back(colTupleList[j]);
}
colOldValueList.push_back(firstPartTupleList);
firstPartTupleList.clear();
}
// end of allocate row id
#ifdef PROFILE
timer.start("writeColumnRec");
#endif
if (rc == NO_ERROR)
{
//----------------------------------------------------------------------
// Mark extents invalid
//----------------------------------------------------------------------
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
if (isFirstBatchPm && (totalRow == rowsLeft))
{
// in this particular case we already marked extents as invalid above.
}
else
{
int firstHalfCount = totalRow - rowsLeft;
for (unsigned i = 0; i < colStructList.size(); i++)
{
colOp = m_colOp[op(colStructList[i].fCompressionType)];
width = colStructList[i].colWidth;
if (firstHalfCount)
{
ExtCPInfo* cpInfoP =
getCPInfoToUpdateForUpdatableType(colStructList[i], &maxMins[i].fSplitMaxMinInfo[0], m_opType);
RID thisRid = rowsLeft ? lastRid : lastRidNew;
successFlag = colOp->calculateRowId(thisRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, colStructList[i], curFbo, cpInfoP));
}
}
maxMins[i].fSplitMaxMinInfoPtrs[0] = cpInfoP;
}
if (rowsLeft)
{
ExtCPInfo* cpInfoP =
getCPInfoToUpdateForUpdatableType(colStructList[i], &maxMins[i].fSplitMaxMinInfo[1], m_opType);
if (cpInfoP)
{
RETURN_ON_ERROR(GetLBIDRange(newExtentsStartingLbids[i], colStructList[i], *cpInfoP));
}
maxMins[i].fSplitMaxMinInfoPtrs[1] = cpInfoP;
}
}
}
markTxnExtentsAsInvalid(txnid);
//----------------------------------------------------------------------
// Write row(s) to database file(s)
//----------------------------------------------------------------------
std::vector<ExtCPInfo> cpinfoList;
for (auto& splitCPInfo : maxMins)
{
for (i = 0; i < 2; i++)
{
ExtCPInfo* cpInfo = splitCPInfo.fSplitMaxMinInfoPtrs[i];
if (cpInfo)
{
cpinfoList.push_back(*cpInfo);
cpinfoList[cpinfoList.size() - 1].fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
}
}
}
rc = BRMWrapper::getInstance()->setExtentsMaxMin(cpinfoList);
if (rc != NO_ERROR)
{
return rc;
}
rc = writeColumnRec(txnid, cscColTypeList, colStructList, colOldValueList, rowIdArray, newColStructList,
colNewValueList, tableOid, useTmpSuffix, true, &maxMins); // @bug 5572 HDFS tmp file
if (rc == NO_ERROR)
{
if (dctnryStructList.size() > 0)
{
vector<BRM::OID_t> oids{static_cast<int32_t>(tableOid)};
for (const DctnryStruct& dctnryStruct : dctnryStructList)
{
oids.push_back(dctnryStruct.dctnryOid);
}
rc = flushOIDsFromCache(oids);
if (rc != 0)
rc = ERR_BLKCACHE_FLUSH_LIST; // translate to WE error
}
if (rc == NO_ERROR)
{
int index = 0;
for (auto& splitCPInfo : maxMins)
{
for (i = 0; i < 2; i++)
{
ExtCPInfo* cpInfo = splitCPInfo.fSplitMaxMinInfoPtrs[i];
if (cpInfo)
{
cpinfoList[index] = *cpInfo;
cpinfoList[index].fCPInfo.seqNum++;
index++;
}
}
}
setInvalidCPInfosSpecialMarks(cpinfoList);
rc = BRMWrapper::getInstance()->setExtentsMaxMin(cpinfoList);
}
}
}
return rc;
}
int WriteEngineWrapper::insertColumnRecsBinary(
const TxnID& txnid, ColStructList& colStructList, std::vector<uint64_t>& colValueList,
DctnryStructList& dctnryStructList, DictStrList& dictStrList,
std::vector<boost::shared_ptr<DBRootExtentTracker> >& dbRootExtentTrackers, RBMetaWriter* fRBMetaWriter,
bool bFirstExtentOnThisPM, bool insertSelect, bool isAutoCommitOn, OID tableOid, bool isFirstBatchPm)
{
int rc;
RID* rowIdArray = NULL;
Column curCol;
ColStruct curColStruct;
ColStructList newColStructList;
std::vector<uint64_t> colNewValueList;
DctnryStructList newDctnryStructList;
HWM hwm = 0;
HWM oldHwm = 0;
HWM newHwm = 0;
size_t totalRow;
ColStructList::size_type totalColumns;
uint64_t rowsLeft = 0;
bool newExtent = false;
RIDList ridList;
ColumnOp* colOp = NULL;
std::vector<BRM::LBID_t> dictLbids;
// Set tmp file suffix to modify HDFS db file
bool useTmpSuffix = false;
m_opType = INSERT;
if (idbdatafile::IDBPolicy::useHdfs())
{
if (!bFirstExtentOnThisPM)
useTmpSuffix = true;
}
unsigned i = 0;
#ifdef PROFILE
StopWatch timer;
#endif
// Convert data type and column width to write engine specific
for (i = 0; i < colStructList.size(); i++)
Convertor::convertColType(&colStructList[i]);
uint32_t colId = 0;
// MCOL-1675: find the smallest column width to calculate the RowID from so
// that all HWMs will be incremented by this operation
findSmallestColumn(colId, colStructList);
// rc = checkValid(txnid, colStructList, colValueList, ridList);
// if (rc != NO_ERROR)
// return rc;
setTransId(txnid);
uint16_t dbRoot, segmentNum;
uint32_t partitionNum;
string segFile;
bool newFile;
TableMetaData* tableMetaData = TableMetaData::makeTableMetaData(tableOid);
// populate colStructList with file information
IDBDataFile* pFile = NULL;
std::vector<DBRootExtentInfo> extentInfo;
int currentDBrootIdx = 0;
std::vector<BRM::CreateStripeColumnExtentsArgOut> extents;
//--------------------------------------------------------------------------
// For first batch on this PM:
// o get starting extent from ExtentTracker, and allocate extent if needed
// o construct colStructList and dctnryStructList accordingly
// o save extent information in tableMetaData for future use
// If not first batch on this PM:
// o construct colStructList and dctnryStructList from tableMetaData
//--------------------------------------------------------------------------
if (isFirstBatchPm)
{
currentDBrootIdx = dbRootExtentTrackers[colId]->getCurrentDBRootIdx();
extentInfo = dbRootExtentTrackers[colId]->getDBRootExtentList();
dbRoot = extentInfo[currentDBrootIdx].fDbRoot;
partitionNum = extentInfo[currentDBrootIdx].fPartition;
//----------------------------------------------------------------------
// check whether this extent is the first on this PM
//----------------------------------------------------------------------
if (bFirstExtentOnThisPM)
{
// cout << "bFirstExtentOnThisPM is " << bFirstExtentOnThisPM << endl;
std::vector<BRM::CreateStripeColumnExtentsArgIn> cols;
BRM::CreateStripeColumnExtentsArgIn createStripeColumnExtentsArgIn;
for (i = 0; i < colStructList.size(); i++)
{
createStripeColumnExtentsArgIn.oid = colStructList[i].dataOid;
createStripeColumnExtentsArgIn.width = colStructList[i].colWidth;
createStripeColumnExtentsArgIn.colDataType = colStructList[i].colDataType;
cols.push_back(createStripeColumnExtentsArgIn);
}
rc = BRMWrapper::getInstance()->allocateStripeColExtents(cols, dbRoot, partitionNum, segmentNum,
extents);
if (rc != NO_ERROR)
return rc;
// Create column files
ExtCPInfoList cpinfoList;
for (i = 0; i < extents.size(); i++)
{
ExtCPInfo cpInfo(colStructList[i].colDataType, colStructList[i].colWidth);
colOp = m_colOp[op(colStructList[i].fCompressionType)];
colOp->initColumn(curCol);
colOp->setColParam(curCol, 0, colStructList[i].colWidth, colStructList[i].colDataType,
colStructList[i].colType, colStructList[i].dataOid,
colStructList[i].fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(colStructList[i].colWidth, colStructList[i].colDataType);
rc = colOp->extendColumn(curCol, false, extents[i].startBlkOffset, extents[i].startLbid,
extents[i].allocSize, dbRoot, partitionNum, segmentNum, segFile, pFile,
newFile);
if (rc != NO_ERROR)
return rc;
cpInfo.toInvalid();
cpInfo.fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
// mark the extents to invalid
cpInfo.fCPInfo.firstLbid = extents[i].startLbid;
cpinfoList.push_back(cpInfo);
colStructList[i].fColPartition = partitionNum;
colStructList[i].fColSegment = segmentNum;
colStructList[i].fColDbRoot = dbRoot;
dctnryStructList[i].fColPartition = partitionNum;
dctnryStructList[i].fColSegment = segmentNum;
dctnryStructList[i].fColDbRoot = dbRoot;
}
// mark the extents to invalid
rc = BRMWrapper::getInstance()->setExtentsMaxMin(cpinfoList);
if (rc != NO_ERROR)
return rc;
// create corresponding dictionary files
for (i = 0; i < dctnryStructList.size(); i++)
{
if (dctnryStructList[i].dctnryOid > 0)
{
rc = createDctnry(txnid, dctnryStructList[i].dctnryOid, dctnryStructList[i].colWidth, dbRoot,
partitionNum, segmentNum, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
return rc;
}
}
} // if ( bFirstExtentOnThisPM)
else // if (!bFirstExtentOnThisPM)
{
std::vector<DBRootExtentInfo> tmpExtentInfo;
for (i = 0; i < dbRootExtentTrackers.size(); i++)
{
tmpExtentInfo = dbRootExtentTrackers[i]->getDBRootExtentList();
colStructList[i].fColPartition = tmpExtentInfo[currentDBrootIdx].fPartition;
colStructList[i].fColSegment = tmpExtentInfo[currentDBrootIdx].fSegment;
colStructList[i].fColDbRoot = tmpExtentInfo[currentDBrootIdx].fDbRoot;
// cout << "Load from dbrootExtenttracker oid:dbroot:part:seg = " <<colStructList[i].dataOid<<":"
//<<colStructList[i].fColDbRoot<<":"<<colStructList[i].fColPartition<<":"<<colStructList[i].fColSegment<<endl;
dctnryStructList[i].fColPartition = tmpExtentInfo[currentDBrootIdx].fPartition;
dctnryStructList[i].fColSegment = tmpExtentInfo[currentDBrootIdx].fSegment;
dctnryStructList[i].fColDbRoot = tmpExtentInfo[currentDBrootIdx].fDbRoot;
}
}
//----------------------------------------------------------------------
// Save the extents info in tableMetaData
//----------------------------------------------------------------------
for (i = 0; i < colStructList.size(); i++)
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = colStructList[i].fColDbRoot;
aExt.partNum = colStructList[i].fColPartition;
aExt.segNum = colStructList[i].fColSegment;
aExt.compType = colStructList[i].fCompressionType;
aExt.isDict = false;
if (bFirstExtentOnThisPM)
{
aExt.hwm = extents[i].startBlkOffset;
aExt.isNewExt = true;
// cout << "adding a ext to metadata" << endl;
}
else
{
std::vector<DBRootExtentInfo> tmpExtentInfo;
tmpExtentInfo = dbRootExtentTrackers[i]->getDBRootExtentList();
aExt.isNewExt = false;
aExt.hwm = tmpExtentInfo[currentDBrootIdx].fLocalHwm;
// cout << "oid " << colStructList[i].dataOid << " gets hwm " << aExt.hwm << endl;
}
aExt.current = true;
aColExtsInfo.push_back(aExt);
// cout << "get from extentinfo oid:hwm = " << colStructList[i].dataOid << ":" << aExt.hwm << endl;
}
tableMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
for (i = 0; i < dctnryStructList.size(); i++)
{
if (dctnryStructList[i].dctnryOid > 0)
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(dctnryStructList[i].dctnryOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == dctnryStructList[i].fColDbRoot) &&
(it->partNum == dctnryStructList[i].fColPartition) &&
(it->segNum == dctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = dctnryStructList[i].fColDbRoot;
aExt.partNum = dctnryStructList[i].fColPartition;
aExt.segNum = dctnryStructList[i].fColSegment;
aExt.compType = dctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
}
tableMetaData->setColExtsInfo(dctnryStructList[i].dctnryOid, aColExtsInfo);
}
}
} // if (isFirstBatchPm)
else // get the extent info from tableMetaData
{
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[colId].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if (it->current)
break;
it++;
}
if (it == aColExtsInfo.end())
return 1;
for (i = 0; i < colStructList.size(); i++)
{
colStructList[i].fColPartition = it->partNum;
colStructList[i].fColSegment = it->segNum;
colStructList[i].fColDbRoot = it->dbRoot;
dctnryStructList[i].fColPartition = it->partNum;
dctnryStructList[i].fColSegment = it->segNum;
dctnryStructList[i].fColDbRoot = it->dbRoot;
}
}
totalColumns = colStructList.size();
totalRow = colValueList.size() / totalColumns;
rowIdArray = new RID[totalRow];
// use scoped_array to ensure ptr deletion regardless of where we return
boost::scoped_array<RID> rowIdArrayPtr(rowIdArray);
memset(rowIdArray, 0, (sizeof(RID) * totalRow));
//--------------------------------------------------------------------------
// allocate row id(s)
//--------------------------------------------------------------------------
curColStruct = colStructList[colId];
colOp = m_colOp[op(curColStruct.fCompressionType)];
colOp->initColumn(curCol);
// Get the correct segment, partition, column file
vector<ExtentInfo> colExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> dictExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> fileInfo;
dbRoot = curColStruct.fColDbRoot;
// use the smallest column to calculate row id
ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[colId].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[colId].fColDbRoot) &&
(it->partNum == colStructList[colId].fColPartition) &&
(it->segNum == colStructList[colId].fColSegment) && it->current)
break;
it++;
}
if (it != aColExtsInfo.end())
{
hwm = it->hwm;
// cout << "Got from colextinfo hwm for oid " << colStructList[colId].dataOid << " is " << hwm << " and
// seg is " << colStructList[colId].fColSegment << endl;
}
oldHwm = hwm; // Save this info for rollback
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, colId, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, curColStruct.fColDbRoot,
curColStruct.fColPartition, curColStruct.fColSegment);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
return rc;
}
// get hwm first
// @bug 286 : fix for bug 286 - correct the typo in getHWM
// RETURN_ON_ERROR(BRMWrapper::getInstance()->getHWM(curColStruct.dataOid, hwm));
Column newCol;
#ifdef PROFILE
timer.start("allocRowId");
#endif
newColStructList = colStructList;
newDctnryStructList = dctnryStructList;
bool bUseStartExtent = true;
if (idbdatafile::IDBPolicy::useHdfs())
insertSelect = true;
rc = colOp->allocRowId(txnid, bUseStartExtent, curCol, (uint64_t)totalRow, rowIdArray, hwm, newExtent,
rowsLeft, newHwm, newFile, newColStructList, newDctnryStructList,
dbRootExtentTrackers, insertSelect, true, tableOid, isFirstBatchPm, NULL);
// cout << "after allocrowid, total row = " <<totalRow << " newExtent is " << newExtent << endl;
// cout << "column oid " << curColStruct.dataOid << " has hwm:newHwm = " << hwm <<":" << newHwm<< endl;
if (rc != NO_ERROR) // Clean up is already done
return rc;
#ifdef PROFILE
timer.stop("allocRowId");
#endif
//--------------------------------------------------------------------------
// Expand initial abbreviated extent if any RID in 1st extent is > 256K.
// if totalRow == rowsLeft, then not adding rows to 1st extent, so skip it.
//--------------------------------------------------------------------------
// DMC-SHARED_NOTHING_NOTE: Is it safe to assume only part0 seg0 is abbreviated?
if ((curCol.dataFile.fPartition == 0) && (curCol.dataFile.fSegment == 0) && ((totalRow - rowsLeft) > 0) &&
(rowIdArray[totalRow - rowsLeft - 1] >= (RID)INITIAL_EXTENT_ROWS_TO_DISK))
{
for (size_t k = 0; k < colStructList.size(); k++)
{
// Skip the selected column
if (k == (size_t)colId)
continue;
Column expandCol;
colOp = m_colOp[op(colStructList[k].fCompressionType)];
// Shouldn't we change 0 to colId here?
colOp->setColParam(expandCol, 0, colStructList[k].colWidth, colStructList[k].colDataType,
colStructList[k].colType, colStructList[k].dataOid,
colStructList[k].fCompressionType, colStructList[k].fColDbRoot,
colStructList[k].fColPartition, colStructList[k].fColSegment);
colOp->findTypeHandler(colStructList[k].colWidth, colStructList[k].colDataType);
rc = colOp->openColumnFile(expandCol, segFile, true); // @bug 5572 HDFS tmp file
if (rc == NO_ERROR)
{
if (colOp->abbreviatedExtent(expandCol.dataFile.pFile, colStructList[k].colWidth))
{
rc = colOp->expandAbbrevExtent(expandCol);
}
}
if (rc != NO_ERROR)
{
return rc;
}
colOp->closeColumnFile(expandCol);
}
}
//--------------------------------------------------------------------------
// Tokenize data if needed
//--------------------------------------------------------------------------
if (insertSelect && isAutoCommitOn)
BRMWrapper::setUseVb(false);
else
BRMWrapper::setUseVb(true);
dictStr::iterator dctStr_iter;
uint64_t* colValPtr;
size_t rowsPerColumn = colValueList.size() / colStructList.size();
for (i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
dctStr_iter = dictStrList[i].begin();
Dctnry* dctnry = m_dctnry[op(dctnryStructList[i].fCompressionType)];
rc = dctnry->openDctnry(dctnryStructList[i].dctnryOid, dctnryStructList[i].fColDbRoot,
dctnryStructList[i].fColPartition, dctnryStructList[i].fColSegment,
useTmpSuffix); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
return rc;
}
for (uint32_t rows = 0; rows < (totalRow - rowsLeft); rows++)
{
colValPtr = &colValueList[(i * rowsPerColumn) + rows];
if (dctStr_iter->isNull())
{
Token nullToken;
memcpy(colValPtr, &nullToken, 8);
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
memcpy(colValPtr, &dctTuple.token, 8);
dictLbids.push_back(dctTuple.token.fbo);
}
dctStr_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry(false);
if (rc != NO_ERROR)
return rc;
if (newExtent)
{
//@Bug 4854 back up hwm chunk for the file to be modified
if (fRBMetaWriter)
fRBMetaWriter->backupDctnryHWMChunk(
newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment);
rc = dctnry->openDctnry(newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment,
false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
for (uint32_t rows = 0; rows < rowsLeft; rows++)
{
colValPtr = &colValueList[(i * rowsPerColumn) + rows];
if (dctStr_iter->isNull())
{
Token nullToken;
memcpy(colValPtr, &nullToken, 8);
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, newDctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
memcpy(colValPtr, &dctTuple.token, 8);
dictLbids.push_back(dctTuple.token.fbo);
}
dctStr_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry(false);
if (rc != NO_ERROR)
return rc;
}
}
}
if (insertSelect && isAutoCommitOn)
BRMWrapper::setUseVb(false);
else
BRMWrapper::setUseVb(true);
//--------------------------------------------------------------------------
// Update column info structure @Bug 1862 set hwm, and
// Prepare ValueList for new extent (if applicable)
//--------------------------------------------------------------------------
//@Bug 2205 Check whether all rows go to the new extent
RID lastRid = 0;
RID lastRidNew = 0;
if (totalRow - rowsLeft > 0)
{
lastRid = rowIdArray[totalRow - rowsLeft - 1];
lastRidNew = rowIdArray[totalRow - 1];
}
else
{
lastRid = 0;
lastRidNew = rowIdArray[totalRow - 1];
}
// cout << "rowid allocated is " << lastRid << endl;
// if a new extent is created, all the columns in this table should have their own new extent
// First column already processed
//@Bug 1701. Close the file (if uncompressed)
m_colOp[op(curCol.compressionType)]->closeColumnFile(curCol);
// cout << "Saving hwm info for new ext batch" << endl;
// Update hwm to set them in the end
bool succFlag = false;
unsigned colWidth = 0;
int curFbo = 0, curBio;
for (i = 0; i < totalColumns; i++)
{
// shoud be obtained from saved hwm
aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[i].dataOid);
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment) && it->current)
break;
it++;
}
if (it != aColExtsInfo.end()) // update hwm info
{
oldHwm = it->hwm;
// save hwm for the old extent
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
// cout << "insertcolumnrec oid:rid:fbo:oldhwm = " << colStructList[i].dataOid << ":" << lastRid <<
// ":" << curFbo << ":" << oldHwm << endl;
if (succFlag)
{
if ((HWM)curFbo >= oldHwm)
{
it->hwm = (HWM)curFbo;
}
//@Bug 4947. set current to false for old extent.
if (newExtent)
{
it->current = false;
}
// cout << "updated old ext info for oid " << colStructList[i].dataOid << "
// dbroot:part:seg:hwm:current = "
//<< it->dbRoot<<":"<<it->partNum<<":"<<it->segNum<<":"<<it->hwm<<":"<< it->current<< " and newExtent
//is " << newExtent << endl;
}
else
return ERR_INVALID_PARAM;
}
// update hwm for the new extent
if (newExtent)
{
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newColStructList[i].fColDbRoot) &&
(it->partNum == newColStructList[i].fColPartition) &&
(it->segNum == newColStructList[i].fColSegment) && it->current)
break;
it++;
}
colWidth = newColStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRidNew, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
if (it != aColExtsInfo.end())
{
it->hwm = (HWM)curFbo;
// cout << "setting hwm to " << (int)curFbo <<" for seg " <<it->segNum << endl;
it->current = true;
}
}
else
return ERR_INVALID_PARAM;
}
tableMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
//--------------------------------------------------------------------------
// Prepare the valuelist for the new extent
//--------------------------------------------------------------------------
for (unsigned i = 1; i <= totalColumns; i++)
{
// Copy values to second value list
for (uint64_t j = rowsLeft; j > 0; j--)
{
colNewValueList.push_back(colValueList[(totalRow * i) - j]);
}
}
// end of allocate row id
#ifdef PROFILE
timer.start("writeColumnRec");
#endif
// cout << "Writing column record" << endl;
if (rc == NO_ERROR)
{
//----------------------------------------------------------------------
// Mark extents invalid
//----------------------------------------------------------------------
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
if (isFirstBatchPm && (totalRow == rowsLeft))
{
}
else
{
for (unsigned i = 0; i < colStructList.size(); i++)
{
colOp = m_colOp[op(colStructList[i].fCompressionType)];
width = colStructList[i].colWidth;
successFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, colStructList[i], curFbo));
}
}
else
return ERR_INVALID_PARAM;
}
}
// If we create a new extent for this batch
for (unsigned i = 0; i < newColStructList.size(); i++)
{
colOp = m_colOp[op(newColStructList[i].fCompressionType)];
width = newColStructList[i].colWidth;
successFlag = colOp->calculateRowId(lastRidNew, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, newColStructList[i], curFbo));
}
}
else
return ERR_INVALID_PARAM;
}
markTxnExtentsAsInvalid(txnid);
//----------------------------------------------------------------------
// Write row(s) to database file(s)
//----------------------------------------------------------------------
bool versioning = !(isAutoCommitOn && insertSelect);
AddDictToList(txnid, dictLbids);
rc =
writeColumnRecBinary(txnid, colStructList, colValueList, rowIdArray, newColStructList,
colNewValueList, tableOid, useTmpSuffix, versioning); // @bug 5572 HDFS tmp file
}
return rc;
}
int WriteEngineWrapper::insertColumnRec_SYS(const TxnID& txnid, const CSCTypesList& cscColTypeList,
ColStructList& colStructList, ColValueList& colValueList,
DctnryStructList& dctnryStructList, DictStrList& dictStrList,
const int32_t tableOid)
{
int rc;
RID* rowIdArray = NULL;
ColTupleList curTupleList;
Column curCol;
ColStruct curColStruct;
ColValueList colOldValueList;
ColValueList colNewValueList;
ColStructList newColStructList;
DctnryStructList newDctnryStructList;
HWM hwm = 0;
HWM newHwm = 0;
HWM oldHwm = 0;
ColTupleList::size_type totalRow;
ColStructList::size_type totalColumns;
uint64_t rowsLeft = 0;
bool newExtent = false;
RIDList ridList;
ColumnOp* colOp = NULL;
uint32_t i = 0;
#ifdef PROFILE
StopWatch timer;
#endif
m_opType = INSERT;
// debug information for testing
if (isDebug(DEBUG_2))
{
printInputValue(colStructList, colValueList, ridList, dctnryStructList, dictStrList);
}
// end
// Convert data type and column width to write engine specific
for (i = 0; i < colStructList.size(); i++)
Convertor::convertColType(&colStructList[i]);
rc = checkValid(txnid, colStructList, colValueList, ridList);
if (rc != NO_ERROR)
return rc;
setTransId(txnid);
curTupleList = static_cast<ColTupleList>(colValueList[0]);
totalRow = curTupleList.size();
totalColumns = colStructList.size();
rowIdArray = new RID[totalRow];
// use scoped_array to ensure ptr deletion regardless of where we return
boost::scoped_array<RID> rowIdArrayPtr(rowIdArray);
memset(rowIdArray, 0, (sizeof(RID) * totalRow));
// allocate row id(s)
curColStruct = colStructList[0];
colOp = m_colOp[op(curColStruct.fCompressionType)];
colOp->initColumn(curCol);
// Get the correct segment, partition, column file
uint16_t dbRoot, segmentNum;
uint32_t partitionNum;
vector<ExtentInfo> colExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> dictExtentInfo; // Save those empty extents in case of failure to rollback
vector<ExtentInfo> fileInfo;
ExtentInfo info;
// Don't search for empty space, always append to the end. May need to revisit this part
dbRoot = curColStruct.fColDbRoot;
int extState;
bool extFound;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getLastHWM_DBroot(curColStruct.dataOid, dbRoot, partitionNum,
segmentNum, hwm, extState, extFound));
for (i = 0; i < colStructList.size(); i++)
{
colStructList[i].fColPartition = partitionNum;
colStructList[i].fColSegment = segmentNum;
colStructList[i].fColDbRoot = dbRoot;
}
oldHwm = hwm; // Save this info for rollback
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, 0, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
string segFile;
rc = colOp->openColumnFile(curCol, segFile, false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
return rc;
}
// get hwm first
// @bug 286 : fix for bug 286 - correct the typo in getHWM
// RETURN_ON_ERROR(BRMWrapper::getInstance()->getHWM(curColStruct.dataOid, hwm));
//...Note that we are casting totalRow to int to be in sync with
//...allocRowId(). So we are assuming that totalRow
//...(curTupleList.size()) will fit into an int. We arleady made
//...that assumption earlier in this method when we used totalRow
//...in the call to calloc() to allocate rowIdArray.
Column newCol;
bool newFile;
#ifdef PROFILE
timer.start("allocRowId");
#endif
newColStructList = colStructList;
newDctnryStructList = dctnryStructList;
std::vector<boost::shared_ptr<DBRootExtentTracker> > dbRootExtentTrackers;
bool bUseStartExtent = true;
rc = colOp->allocRowId(txnid, bUseStartExtent, curCol, (uint64_t)totalRow, rowIdArray, hwm, newExtent,
rowsLeft, newHwm, newFile, newColStructList, newDctnryStructList,
dbRootExtentTrackers, false, false, 0, false, NULL);
if ((rc == ERR_FILE_DISK_SPACE) && newExtent)
{
for (i = 0; i < newColStructList.size(); i++)
{
info.oid = newColStructList[i].dataOid;
info.partitionNum = newColStructList[i].fColPartition;
info.segmentNum = newColStructList[i].fColSegment;
info.dbRoot = newColStructList[i].fColDbRoot;
if (newFile)
fileInfo.push_back(info);
colExtentInfo.push_back(info);
}
int rc1 = BRMWrapper::getInstance()->deleteEmptyColExtents(colExtentInfo);
if ((rc1 == 0) && newFile)
{
rc1 = colOp->deleteFile(fileInfo[0].oid, fileInfo[0].dbRoot, fileInfo[0].partitionNum,
fileInfo[0].segmentNum);
if (rc1 != NO_ERROR)
return rc;
FileOp fileOp;
for (i = 0; i < newDctnryStructList.size(); i++)
{
if (newDctnryStructList[i].dctnryOid > 0)
{
info.oid = newDctnryStructList[i].dctnryOid;
info.partitionNum = newDctnryStructList[i].fColPartition;
info.segmentNum = newDctnryStructList[i].fColSegment;
info.dbRoot = newDctnryStructList[i].fColDbRoot;
info.newFile = true;
fileInfo.push_back(info);
dictExtentInfo.push_back(info);
}
}
if (dictExtentInfo.size() > 0)
{
rc1 = BRMWrapper::getInstance()->deleteEmptyDictStoreExtents(dictExtentInfo);
if (rc1 != NO_ERROR)
return rc;
for (unsigned j = 0; j < fileInfo.size(); j++)
{
rc1 = fileOp.deleteFile(fileInfo[j].oid, fileInfo[j].dbRoot, fileInfo[j].partitionNum,
fileInfo[j].segmentNum);
}
}
}
}
TableMetaData* aTableMetaData = TableMetaData::makeTableMetaData(tableOid);
//..Expand initial abbreviated extent if any RID in 1st extent is > 256K
// DMC-SHARED_NOTHING_NOTE: Is it safe to assume only part0 seg0 is abbreviated?
if ((partitionNum == 0) && (segmentNum == 0) && ((totalRow - rowsLeft) > 0) &&
(rowIdArray[totalRow - rowsLeft - 1] >= (RID)INITIAL_EXTENT_ROWS_TO_DISK))
{
for (size_t k = 1; k < colStructList.size(); k++)
{
Column expandCol;
colOp = m_colOp[op(colStructList[k].fCompressionType)];
colOp->setColParam(expandCol, 0, colStructList[k].colWidth, colStructList[k].colDataType,
colStructList[k].colType, colStructList[k].dataOid,
colStructList[k].fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(colStructList[k].colWidth, colStructList[k].colDataType);
rc = colOp->openColumnFile(expandCol, segFile, false); // @bug 5572 HDFS tmp file
if (rc == NO_ERROR)
{
if (colOp->abbreviatedExtent(expandCol.dataFile.pFile, colStructList[k].colWidth))
{
rc = colOp->expandAbbrevExtent(expandCol);
}
}
if (rc != NO_ERROR)
{
if (newExtent)
{
// Remove the empty extent added to the first column
int rc1 = BRMWrapper::getInstance()->deleteEmptyColExtents(colExtentInfo);
if ((rc1 == 0) && newFile)
{
rc1 = colOp->deleteFile(fileInfo[0].oid, fileInfo[0].dbRoot, fileInfo[0].partitionNum,
fileInfo[0].segmentNum);
}
}
colOp->clearColumn(expandCol); // closes the file
return rc;
}
colOp->clearColumn(expandCol); // closes the file
}
}
BRMWrapper::setUseVb(true);
// Tokenize data if needed
dictStr::iterator dctStr_iter;
ColTupleList::iterator col_iter;
for (i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
dctStr_iter = dictStrList[i].begin();
col_iter = colValueList[i].begin();
Dctnry* dctnry = m_dctnry[op(dctnryStructList[i].fCompressionType)];
dctnryStructList[i].fColPartition = partitionNum;
dctnryStructList[i].fColSegment = segmentNum;
dctnryStructList[i].fColDbRoot = dbRoot;
rc = dctnry->openDctnry(dctnryStructList[i].dctnryOid, dctnryStructList[i].fColDbRoot,
dctnryStructList[i].fColPartition, dctnryStructList[i].fColSegment,
false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
ColExtsInfo aColExtsInfo = aTableMetaData->getColExtsInfo(dctnryStructList[i].dctnryOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == dctnryStructList[i].fColDbRoot) &&
(it->partNum == dctnryStructList[i].fColPartition) &&
(it->segNum == dctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = dctnryStructList[i].fColDbRoot;
aExt.partNum = dctnryStructList[i].fColPartition;
aExt.segNum = dctnryStructList[i].fColSegment;
aExt.compType = dctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
aTableMetaData->setColExtsInfo(dctnryStructList[i].dctnryOid, aColExtsInfo);
}
for (uint32_t rows = 0; rows < (totalRow - rowsLeft); rows++)
{
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry();
if (rc != NO_ERROR)
return rc;
if (newExtent)
{
rc = dctnry->openDctnry(newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment,
false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
aColExtsInfo = aTableMetaData->getColExtsInfo(newDctnryStructList[i].dctnryOid);
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newDctnryStructList[i].fColDbRoot) &&
(it->partNum == newDctnryStructList[i].fColPartition) &&
(it->segNum == newDctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = newDctnryStructList[i].fColDbRoot;
aExt.partNum = newDctnryStructList[i].fColPartition;
aExt.segNum = newDctnryStructList[i].fColSegment;
aExt.compType = newDctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
aTableMetaData->setColExtsInfo(newDctnryStructList[i].dctnryOid, aColExtsInfo);
}
for (uint32_t rows = 0; rows < rowsLeft; rows++)
{
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, newDctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry();
if (rc != NO_ERROR)
return rc;
}
}
}
// Update column info structure @Bug 1862 set hwm
//@Bug 2205 Check whether all rows go to the new extent
RID lastRid = 0;
RID lastRidNew = 0;
if (totalRow - rowsLeft > 0)
{
lastRid = rowIdArray[totalRow - rowsLeft - 1];
lastRidNew = rowIdArray[totalRow - 1];
}
else
{
lastRid = 0;
lastRidNew = rowIdArray[totalRow - 1];
}
// cout << "rowid allocated is " << lastRid << endl;
// if a new extent is created, all the columns in this table should have their own new extent
//@Bug 1701. Close the file
m_colOp[op(curCol.compressionType)]->clearColumn(curCol);
std::vector<BulkSetHWMArg> hwmVecNewext;
std::vector<BulkSetHWMArg> hwmVecOldext;
if (newExtent) // Save all hwms to set them later.
{
BulkSetHWMArg aHwmEntryNew;
BulkSetHWMArg aHwmEntryOld;
bool succFlag = false;
unsigned colWidth = 0;
int extState;
bool extFound;
int curFbo = 0, curBio;
for (i = 0; i < totalColumns; i++)
{
Column curColLocal;
colOp->initColumn(curColLocal);
colOp = m_colOp[op(newColStructList[i].fCompressionType)];
colOp->setColParam(curColLocal, 0, newColStructList[i].colWidth, newColStructList[i].colDataType,
newColStructList[i].colType, newColStructList[i].dataOid,
newColStructList[i].fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(newColStructList[i].colWidth, newColStructList[i].colDataType);
rc = BRMWrapper::getInstance()->getLastHWM_DBroot(curColLocal.dataFile.fid, dbRoot, partitionNum,
segmentNum, oldHwm, extState, extFound);
info.oid = curColLocal.dataFile.fid;
info.partitionNum = partitionNum;
info.segmentNum = segmentNum;
info.dbRoot = dbRoot;
info.hwm = oldHwm;
colExtentInfo.push_back(info);
// @Bug 2714 need to set hwm for the old extent
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
// cout << "insertcolumnrec oid:rid:fbo:hwm = " << colStructList[i].dataOid << ":" << lastRid << ":"
// << curFbo << ":" << hwm << endl;
if (succFlag)
{
if ((HWM)curFbo > oldHwm)
{
aHwmEntryOld.oid = colStructList[i].dataOid;
aHwmEntryOld.partNum = partitionNum;
aHwmEntryOld.segNum = segmentNum;
aHwmEntryOld.hwm = curFbo;
hwmVecOldext.push_back(aHwmEntryOld);
}
}
else
return ERR_INVALID_PARAM;
colWidth = newColStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRidNew, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
aHwmEntryNew.oid = newColStructList[i].dataOid;
aHwmEntryNew.partNum = newColStructList[i].fColPartition;
aHwmEntryNew.segNum = newColStructList[i].fColSegment;
aHwmEntryNew.hwm = curFbo;
hwmVecNewext.push_back(aHwmEntryNew);
}
m_colOp[op(curColLocal.compressionType)]->clearColumn(curColLocal);
}
// Prepare the valuelist for the new extent
ColTupleList colTupleList;
ColTupleList newColTupleList;
ColTupleList firstPartTupleList;
for (unsigned i = 0; i < totalColumns; i++)
{
colTupleList = static_cast<ColTupleList>(colValueList[i]);
for (uint64_t j = rowsLeft; j > 0; j--)
{
newColTupleList.push_back(colTupleList[totalRow - j]);
}
colNewValueList.push_back(newColTupleList);
newColTupleList.clear();
// upate the oldvalue list for the old extent
for (uint64_t j = 0; j < (totalRow - rowsLeft); j++)
{
firstPartTupleList.push_back(colTupleList[j]);
}
colOldValueList.push_back(firstPartTupleList);
firstPartTupleList.clear();
}
}
// Mark extents invalid
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
if (totalRow - rowsLeft > 0)
{
for (unsigned i = 0; i < colStructList.size(); i++)
{
colOp = m_colOp[op(colStructList[i].fCompressionType)];
width = colStructList[i].colWidth;
successFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, colStructList[i], curFbo));
}
}
}
}
lastRid = rowIdArray[totalRow - 1];
for (unsigned i = 0; i < newColStructList.size(); i++)
{
colOp = m_colOp[op(newColStructList[i].fCompressionType)];
width = newColStructList[i].colWidth;
successFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, newColStructList[i], curFbo));
}
}
}
// cout << "lbids size = " << lbids.size()<< endl;
markTxnExtentsAsInvalid(txnid);
if (rc == NO_ERROR)
{
// MCOL-66 The DBRM can't handle concurrent transactions to sys tables
static boost::mutex dbrmMutex;
boost::mutex::scoped_lock lk(dbrmMutex);
if (newExtent)
{
rc = writeColumnRec(txnid, cscColTypeList, colStructList, colOldValueList, rowIdArray, newColStructList,
colNewValueList, tableOid, false); // @bug 5572 HDFS tmp file
}
else
{
rc = writeColumnRec(txnid, cscColTypeList, colStructList, colValueList, rowIdArray, newColStructList,
colNewValueList, tableOid, false); // @bug 5572 HDFS tmp file
}
}
#ifdef PROFILE
timer.stop("writeColumnRec");
#endif
// for (ColTupleList::size_type i = 0; i < totalRow; i++)
// ridList.push_back((RID) rowIdArray[i]);
// if (rc == NO_ERROR)
// rc = flushDataFiles(NO_ERROR);
if (!newExtent)
{
// flushVMCache();
bool succFlag = false;
unsigned colWidth = 0;
int extState;
bool extFound;
int curFbo = 0, curBio;
std::vector<BulkSetHWMArg> hwmVec;
for (unsigned i = 0; i < totalColumns; i++)
{
// colOp = m_colOp[op(colStructList[i].fCompressionType)];
// Set all columns hwm together
BulkSetHWMArg aHwmEntry;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getLastHWM_DBroot(
colStructList[i].dataOid, dbRoot, partitionNum, segmentNum, hwm, extState, extFound));
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
// cout << "insertcolumnrec oid:rid:fbo:hwm = " << colStructList[i].dataOid << ":" << lastRid << ":"
// << curFbo << ":" << hwm << endl;
if (succFlag)
{
if ((HWM)curFbo > hwm)
{
aHwmEntry.oid = colStructList[i].dataOid;
aHwmEntry.partNum = partitionNum;
aHwmEntry.segNum = segmentNum;
aHwmEntry.hwm = curFbo;
hwmVec.push_back(aHwmEntry);
}
}
else
return ERR_INVALID_PARAM;
}
if (hwmVec.size() > 0)
{
std::vector<BRM::CPInfoMerge> mergeCPDataArgs;
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVec, mergeCPDataArgs));
}
}
if (newExtent)
{
#ifdef PROFILE
timer.start("flushVMCache");
#endif
std::vector<BRM::CPInfoMerge> mergeCPDataArgs;
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVecNewext, mergeCPDataArgs));
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVecOldext, mergeCPDataArgs));
// flushVMCache();
#ifdef PROFILE
timer.stop("flushVMCache");
#endif
}
#ifdef PROFILE
timer.finish();
#endif
return rc;
}
int WriteEngineWrapper::insertColumnRec_Single(const TxnID& txnid, const CSCTypesList& cscColTypeList,
ColStructList& colStructList, ColValueList& colValueList,
DctnryStructList& dctnryStructList, DictStrList& dictStrList,
const int32_t tableOid)
{
int rc;
RID* rowIdArray = NULL;
ColTupleList curTupleList;
Column curCol;
ColStruct curColStruct;
ColValueList colOldValueList;
ColValueList colNewValueList;
ColStructList newColStructList;
DctnryStructList newDctnryStructList;
HWM hwm = 0;
HWM newHwm = 0;
HWM oldHwm = 0;
ColTupleList::size_type totalRow;
ColStructList::size_type totalColumns;
uint64_t rowsLeft = 0;
bool newExtent = false;
RIDList ridList;
ColumnOp* colOp = NULL;
uint32_t i = 0;
#ifdef PROFILE
StopWatch timer;
#endif
m_opType = INSERT;
// debug information for testing
if (isDebug(DEBUG_2))
{
printInputValue(colStructList, colValueList, ridList, dctnryStructList, dictStrList);
}
// Convert data type and column width to write engine specific
for (i = 0; i < colStructList.size(); i++)
Convertor::convertColType(&colStructList[i]);
uint32_t colId = 0;
// MCOL-1675: find the smallest column width to calculate the RowID from so
// that all HWMs will be incremented by this operation
findSmallestColumn(colId, colStructList);
rc = checkValid(txnid, colStructList, colValueList, ridList);
if (rc != NO_ERROR)
return rc;
setTransId(txnid);
curTupleList = static_cast<ColTupleList>(colValueList[0]);
totalRow = curTupleList.size();
totalColumns = colStructList.size();
rowIdArray = new RID[totalRow];
// use scoped_array to ensure ptr deletion regardless of where we return
boost::scoped_array<RID> rowIdArrayPtr(rowIdArray);
memset(rowIdArray, 0, (sizeof(RID) * totalRow));
//--------------------------------------------------------------------------
// allocate row id(s)
//--------------------------------------------------------------------------
curColStruct = colStructList[colId];
colOp = m_colOp[op(curColStruct.fCompressionType)];
colOp->initColumn(curCol);
// Get the correct segment, partition, column file
uint16_t dbRoot;
uint16_t segmentNum = 0;
uint32_t partitionNum = 0;
// Don't search for empty space, always append to the end. May revisit later
dbRoot = curColStruct.fColDbRoot;
int extState;
bool bStartExtFound;
bool bUseStartExtent = false;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getLastHWM_DBroot(curColStruct.dataOid, dbRoot, partitionNum,
segmentNum, hwm, extState, bStartExtFound));
if ((bStartExtFound) && (extState == BRM::EXTENTAVAILABLE))
bUseStartExtent = true;
for (i = 0; i < colStructList.size(); i++)
{
colStructList[i].fColPartition = partitionNum;
colStructList[i].fColSegment = segmentNum;
colStructList[i].fColDbRoot = dbRoot;
}
for (i = 0; i < dctnryStructList.size(); i++)
{
dctnryStructList[i].fColPartition = partitionNum;
dctnryStructList[i].fColSegment = segmentNum;
dctnryStructList[i].fColDbRoot = dbRoot;
}
oldHwm = hwm; // Save this info for rollback
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, colId, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, dbRoot, partitionNum, segmentNum);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
string segFile;
if (bUseStartExtent)
{
rc = colOp->openColumnFile(curCol, segFile, true); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
{
return rc;
}
}
bool newFile;
#ifdef PROFILE
timer.start("allocRowId");
#endif
newColStructList = colStructList;
newDctnryStructList = dctnryStructList;
std::vector<boost::shared_ptr<DBRootExtentTracker> > dbRootExtentTrackers;
rc = colOp->allocRowId(txnid, bUseStartExtent, curCol, (uint64_t)totalRow, rowIdArray, hwm, newExtent,
rowsLeft, newHwm, newFile, newColStructList, newDctnryStructList,
dbRootExtentTrackers, false, false, 0, false, NULL);
//--------------------------------------------------------------------------
// Handle case where we ran out of disk space allocating a new extent.
// Rollback extentmap and delete any db files that were created.
//--------------------------------------------------------------------------
if (rc != NO_ERROR)
{
if ((rc == ERR_FILE_DISK_SPACE) && newExtent)
{
vector<ExtentInfo> colExtentInfo;
vector<ExtentInfo> dictExtentInfo;
vector<ExtentInfo> fileInfo;
ExtentInfo info;
for (i = 0; i < newColStructList.size(); i++)
{
info.oid = newColStructList[i].dataOid;
info.partitionNum = newColStructList[i].fColPartition;
info.segmentNum = newColStructList[i].fColSegment;
info.dbRoot = newColStructList[i].fColDbRoot;
if (newFile)
fileInfo.push_back(info);
colExtentInfo.push_back(info);
}
int rc1 = BRMWrapper::getInstance()->deleteEmptyColExtents(colExtentInfo);
// Only rollback dictionary extents "if" store file is new
if ((rc1 == 0) && newFile)
{
for (unsigned int j = 0; j < fileInfo.size(); j++)
{
// ignore return code and delete what we can
rc1 = colOp->deleteFile(fileInfo[j].oid, fileInfo[j].dbRoot, fileInfo[j].partitionNum,
fileInfo[j].segmentNum);
}
fileInfo.clear();
for (i = 0; i < newDctnryStructList.size(); i++)
{
if (newDctnryStructList[i].dctnryOid > 0)
{
info.oid = newDctnryStructList[i].dctnryOid;
info.partitionNum = newDctnryStructList[i].fColPartition;
info.segmentNum = newDctnryStructList[i].fColSegment;
info.dbRoot = newDctnryStructList[i].fColDbRoot;
info.newFile = true;
fileInfo.push_back(info);
dictExtentInfo.push_back(info);
}
}
if (dictExtentInfo.size() > 0)
{
FileOp fileOp;
rc1 = BRMWrapper::getInstance()->deleteEmptyDictStoreExtents(dictExtentInfo);
if (rc1 != NO_ERROR)
return rc;
for (unsigned j = 0; j < fileInfo.size(); j++)
{
rc1 = fileOp.deleteFile(fileInfo[j].oid, fileInfo[j].dbRoot, fileInfo[j].partitionNum,
fileInfo[j].segmentNum);
}
}
}
} // disk space error allocating new extent
return rc;
} // rc != NO_ERROR from call to allocRowID()
#ifdef PROFILE
timer.stop("allocRowId");
#endif
TableMetaData* aTableMetaData = TableMetaData::makeTableMetaData(tableOid);
//--------------------------------------------------------------------------
// Expand initial abbreviated extent if any RID in 1st extent is > 256K.
// if totalRow == rowsLeft, then not adding rows to 1st extent, so skip it.
//--------------------------------------------------------------------------
// DMC-SHARED_NOTHING_NOTE: Is it safe to assume only part0 seg0 is abbreviated?
if ((colStructList[colId].fColPartition == 0) && (colStructList[colId].fColSegment == 0) &&
((totalRow - rowsLeft) > 0) &&
(rowIdArray[totalRow - rowsLeft - 1] >= (RID)INITIAL_EXTENT_ROWS_TO_DISK))
{
for (unsigned k = 0; k < colStructList.size(); k++)
{
if (k == colId)
continue;
Column expandCol;
colOp = m_colOp[op(colStructList[k].fCompressionType)];
colOp->setColParam(expandCol, 0, colStructList[k].colWidth, colStructList[k].colDataType,
colStructList[k].colType, colStructList[k].dataOid,
colStructList[k].fCompressionType, colStructList[k].fColDbRoot,
colStructList[k].fColPartition, colStructList[k].fColSegment);
colOp->findTypeHandler(colStructList[k].colWidth, colStructList[k].colDataType);
rc = colOp->openColumnFile(expandCol, segFile, true); // @bug 5572 HDFS tmp file
if (rc == NO_ERROR)
{
if (colOp->abbreviatedExtent(expandCol.dataFile.pFile, colStructList[k].colWidth))
{
rc = colOp->expandAbbrevExtent(expandCol);
}
}
colOp->clearColumn(expandCol); // closes the file
if (rc != NO_ERROR)
{
return rc;
}
} // loop through columns
} // if starting extent needs to be expanded
//--------------------------------------------------------------------------
// Tokenize data if needed
//--------------------------------------------------------------------------
dictStr::iterator dctStr_iter;
ColTupleList::iterator col_iter;
for (unsigned i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
dctStr_iter = dictStrList[i].begin();
col_iter = colValueList[i].begin();
Dctnry* dctnry = m_dctnry[op(dctnryStructList[i].fCompressionType)];
ColExtsInfo aColExtsInfo = aTableMetaData->getColExtsInfo(dctnryStructList[i].dctnryOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
if (bUseStartExtent)
{
rc = dctnry->openDctnry(dctnryStructList[i].dctnryOid, dctnryStructList[i].fColDbRoot,
dctnryStructList[i].fColPartition, dctnryStructList[i].fColSegment,
true); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == dctnryStructList[i].fColDbRoot) &&
(it->partNum == dctnryStructList[i].fColPartition) &&
(it->segNum == dctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = dctnryStructList[i].fColDbRoot;
aExt.partNum = dctnryStructList[i].fColPartition;
aExt.segNum = dctnryStructList[i].fColSegment;
aExt.compType = dctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
aTableMetaData->setColExtsInfo(dctnryStructList[i].dctnryOid, aColExtsInfo);
}
for (uint32_t rows = 0; rows < (totalRow - rowsLeft); rows++)
{
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, dctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry();
if (rc != NO_ERROR)
return rc;
} // tokenize dictionary rows in 1st extent
if (newExtent)
{
rc = dctnry->openDctnry(newDctnryStructList[i].dctnryOid, newDctnryStructList[i].fColDbRoot,
newDctnryStructList[i].fColPartition, newDctnryStructList[i].fColSegment,
false); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
return rc;
aColExtsInfo = aTableMetaData->getColExtsInfo(newDctnryStructList[i].dctnryOid);
it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newDctnryStructList[i].fColDbRoot) &&
(it->partNum == newDctnryStructList[i].fColPartition) &&
(it->segNum == newDctnryStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = newDctnryStructList[i].fColDbRoot;
aExt.partNum = newDctnryStructList[i].fColPartition;
aExt.segNum = newDctnryStructList[i].fColSegment;
aExt.compType = newDctnryStructList[i].fCompressionType;
aExt.isDict = true;
aColExtsInfo.push_back(aExt);
aTableMetaData->setColExtsInfo(newDctnryStructList[i].dctnryOid, aColExtsInfo);
}
for (uint32_t rows = 0; rows < rowsLeft; rows++)
{
if (dctStr_iter->isNull())
{
Token nullToken;
col_iter->data = nullToken;
}
else
{
#ifdef PROFILE
timer.start("tokenize");
#endif
DctnryTuple dctTuple;
dctTuple.sigValue = (unsigned char*)dctStr_iter->str();
dctTuple.sigSize = dctStr_iter->length();
dctTuple.isNull = false;
rc = tokenize(txnid, dctTuple, newDctnryStructList[i].fCompressionType);
if (rc != NO_ERROR)
{
dctnry->closeDctnry();
return rc;
}
#ifdef PROFILE
timer.stop("tokenize");
#endif
col_iter->data = dctTuple.token;
}
dctStr_iter++;
col_iter++;
}
// close dictionary files
rc = dctnry->closeDctnry();
if (rc != NO_ERROR)
return rc;
} // tokenize dictionary rows in second extent
} // tokenize dictionary columns
} // loop through columns to see which ones need tokenizing
//----------------------------------------------------------------------
// Update column info structure @Bug 1862 set hwm, and
// Prepare ValueList for new extent (if applicable)
//----------------------------------------------------------------------
//@Bug 2205 Check whether all rows go to the new extent
RID lastRid = 0;
RID lastRidNew = 0;
if (totalRow - rowsLeft > 0)
{
lastRid = rowIdArray[totalRow - rowsLeft - 1];
lastRidNew = rowIdArray[totalRow - 1];
}
else
{
lastRid = 0;
lastRidNew = rowIdArray[totalRow - 1];
}
// if a new extent is created, all the columns in this table should
// have their own new extent
//@Bug 1701. Close the file
if (bUseStartExtent)
{
m_colOp[op(curCol.compressionType)]->clearColumn(curCol);
}
std::vector<BulkSetHWMArg> hwmVecNewext;
std::vector<BulkSetHWMArg> hwmVecOldext;
if (newExtent) // Save all hwms to set them later.
{
BulkSetHWMArg aHwmEntryNew;
BulkSetHWMArg aHwmEntryOld;
bool succFlag = false;
unsigned colWidth = 0;
int curFbo = 0, curBio;
for (i = 0; i < totalColumns; i++)
{
colOp = m_colOp[op(newColStructList[i].fCompressionType)];
// @Bug 2714 need to set hwm for the old extent
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
if ((HWM)curFbo > oldHwm)
{
aHwmEntryOld.oid = colStructList[i].dataOid;
aHwmEntryOld.partNum = colStructList[i].fColPartition;
aHwmEntryOld.segNum = colStructList[i].fColSegment;
aHwmEntryOld.hwm = curFbo;
hwmVecOldext.push_back(aHwmEntryOld);
}
}
else
return ERR_INVALID_PARAM;
colWidth = newColStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRidNew, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
if (succFlag)
{
aHwmEntryNew.oid = newColStructList[i].dataOid;
aHwmEntryNew.partNum = newColStructList[i].fColPartition;
aHwmEntryNew.segNum = newColStructList[i].fColSegment;
aHwmEntryNew.hwm = curFbo;
hwmVecNewext.push_back(aHwmEntryNew);
}
}
//----------------------------------------------------------------------
// Prepare the valuelist for the new extent
//----------------------------------------------------------------------
ColTupleList colTupleList;
ColTupleList newColTupleList;
ColTupleList firstPartTupleList;
for (unsigned i = 0; i < totalColumns; i++)
{
colTupleList = static_cast<ColTupleList>(colValueList[i]);
for (uint64_t j = rowsLeft; j > 0; j--)
{
newColTupleList.push_back(colTupleList[totalRow - j]);
}
colNewValueList.push_back(newColTupleList);
newColTupleList.clear();
// upate the oldvalue list for the old extent
for (uint64_t j = 0; j < (totalRow - rowsLeft); j++)
{
firstPartTupleList.push_back(colTupleList[j]);
}
colOldValueList.push_back(firstPartTupleList);
firstPartTupleList.clear();
}
}
//--------------------------------------------------------------------------
// Mark extents invalid
//--------------------------------------------------------------------------
// WIP
// Set min/max in dmlprocprocessor if aplicable
vector<BRM::LBID_t> lbids;
vector<CalpontSystemCatalog::ColDataType> colDataTypes;
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
lastRid = rowIdArray[totalRow - 1];
for (unsigned i = 0; i < colStructList.size(); i++)
{
colOp = m_colOp[op(colStructList[i].fCompressionType)];
width = colStructList[i].colWidth;
successFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
colDataTypes.push_back(colStructList[i].colDataType);
RETURN_ON_ERROR(AddLBIDtoList(txnid, colStructList[i], curFbo));
}
}
}
markTxnExtentsAsInvalid(txnid);
//--------------------------------------------------------------------------
// Write row(s) to database file(s)
//--------------------------------------------------------------------------
#ifdef PROFILE
timer.start("writeColumnRec");
#endif
if (rc == NO_ERROR)
{
if (newExtent)
{
rc = writeColumnRec(txnid, cscColTypeList, colStructList, colOldValueList, rowIdArray, newColStructList,
colNewValueList, tableOid,
false); // @bug 5572 HDFS tmp file
}
else
{
rc = writeColumnRec(txnid, cscColTypeList, colStructList, colValueList, rowIdArray, newColStructList,
colNewValueList, tableOid,
true); // @bug 5572 HDFS tmp file
}
}
#ifdef PROFILE
timer.stop("writeColumnRec");
#endif
//--------------------------------------------------------------------------
// Update BRM
//--------------------------------------------------------------------------
if (!newExtent)
{
bool succFlag = false;
unsigned colWidth = 0;
int extState;
bool extFound;
int curFbo = 0, curBio;
std::vector<BulkSetHWMArg> hwmVec;
for (unsigned i = 0; i < totalColumns; i++)
{
// Set all columns hwm together
BulkSetHWMArg aHwmEntry;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getLastHWM_DBroot(
colStructList[i].dataOid, colStructList[i].fColDbRoot, colStructList[i].fColPartition,
colStructList[i].fColSegment, hwm, extState, extFound));
colWidth = colStructList[i].colWidth;
succFlag = colOp->calculateRowId(lastRid, BYTE_PER_BLOCK / colWidth, colWidth, curFbo, curBio);
// cout << "insertcolumnrec oid:rid:fbo:hwm = " <<
// colStructList[i].dataOid << ":" << lastRid << ":" <<
// curFbo << ":" << hwm << endl;
if (succFlag)
{
if ((HWM)curFbo > hwm)
{
aHwmEntry.oid = colStructList[i].dataOid;
aHwmEntry.partNum = colStructList[i].fColPartition;
aHwmEntry.segNum = colStructList[i].fColSegment;
aHwmEntry.hwm = curFbo;
hwmVec.push_back(aHwmEntry);
}
}
else
return ERR_INVALID_PARAM;
}
std::vector<BRM::CPInfoMerge> mergeCPDataArgs;
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVec, mergeCPDataArgs));
}
else // if (newExtent)
{
#ifdef PROFILE
timer.start("flushVMCache");
#endif
std::vector<BRM::CPInfoMerge> mergeCPDataArgs;
if (hwmVecNewext.size() > 0)
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVecNewext, mergeCPDataArgs));
if (hwmVecOldext.size() > 0)
RETURN_ON_ERROR(BRMWrapper::getInstance()->bulkSetHWMAndCP(hwmVecOldext, mergeCPDataArgs));
#ifdef PROFILE
timer.stop("flushVMCache");
#endif
}
#ifdef PROFILE
timer.finish();
#endif
return rc;
}
/*@brief printInputValue - Print input value
*/
/***********************************************************
* DESCRIPTION:
* Print input value
* PARAMETERS:
* tableOid - table object id
* colStructList - column struct list
* colValueList - column value list
* ridList - RID list
* RETURN:
* none
***********************************************************/
void WriteEngineWrapper::printInputValue(const ColStructList& colStructList, const ColValueList& colValueList,
const RIDList& ridList, const DctnryStructList& dctnryStructList,
const DictStrList& dictStrList) const
{
ColTupleList curTupleList;
ColStruct curColStruct;
ColTuple curTuple;
string curStr;
ColStructList::size_type i;
size_t j;
OidToIdxMap oidToIdxMap;
std::cerr << std::endl << "=========================" << std::endl;
std::cerr << "Total RIDs: " << ridList.size() << std::endl;
for (i = 0; i < ridList.size(); i++)
std::cerr << "RID[" << i << "] : " << ridList[i] << std::endl;
std::cerr << "Total Columns: " << colStructList.size() << std::endl;
for (i = 0; i < colStructList.size(); i++)
{
curColStruct = colStructList[i];
curTupleList = colValueList[i];
if (curColStruct.tokenFlag)
{
oidToIdxMap.insert({curColStruct.dataOid, i});
continue;
}
std::cerr << "Column[" << i << "]";
std::cerr << "Data file OID : " << curColStruct.dataOid << "\t";
std::cerr << "Width : " << curColStruct.colWidth << "\t"
<< " Type: " << curColStruct.colDataType << std::endl;
std::cerr << "Total values : " << curTupleList.size() << std::endl;
for (j = 0; j < curTupleList.size(); j++)
{
curTuple = curTupleList[j];
try
{
if (curTuple.data.type() == typeid(int))
curStr = boost::lexical_cast<string>(boost::any_cast<int>(curTuple.data));
else if (curTuple.data.type() == typeid(unsigned int))
curStr = boost::lexical_cast<string>(boost::any_cast<unsigned int>(curTuple.data));
else if (curTuple.data.type() == typeid(float))
curStr = boost::lexical_cast<string>(boost::any_cast<float>(curTuple.data));
else if (curTuple.data.type() == typeid(long long))
curStr = boost::lexical_cast<string>(boost::any_cast<long long>(curTuple.data));
else if (curTuple.data.type() == typeid(unsigned long long))
curStr = boost::lexical_cast<string>(boost::any_cast<unsigned long long>(curTuple.data));
else if (curTuple.data.type() == typeid(int128_t))
curStr = datatypes::TSInt128(boost::any_cast<int128_t>(curTuple.data)).toString();
else if (curTuple.data.type() == typeid(double))
curStr = boost::lexical_cast<string>(boost::any_cast<double>(curTuple.data));
else if (curTuple.data.type() == typeid(short))
curStr = boost::lexical_cast<string>(boost::any_cast<short>(curTuple.data));
else if (curTuple.data.type() == typeid(unsigned short))
curStr = boost::lexical_cast<string>(boost::any_cast<unsigned short>(curTuple.data));
else if (curTuple.data.type() == typeid(char))
curStr = boost::lexical_cast<string>(boost::any_cast<char>(curTuple.data));
else
curStr = boost::any_cast<string>(curTuple.data);
}
catch (...)
{
}
if (isDebug(DEBUG_3))
std::cerr << "Value[" << j << "]: " << curStr.c_str() << std::endl;
}
}
for (i = 0; i < dctnryStructList.size(); ++i)
{
if (dctnryStructList[i].dctnryOid == 0)
continue;
std::cerr << "Dict[" << i << "]";
std::cerr << " file OID : " << dctnryStructList[i].dctnryOid
<< " Token file OID: " << dctnryStructList[i].columnOid << "\t";
std::cerr << "Width : " << dctnryStructList[i].colWidth << "\t"
<< " Type: " << dctnryStructList[i].fCompressionType << std::endl;
std::cerr << "Total values : " << dictStrList.size() << std::endl;
if (isDebug(DEBUG_3))
{
for (j = 0; j < dictStrList[i].size(); ++j)
{
// We presume there will be a value.
auto tokenOidIdx = oidToIdxMap[dctnryStructList[i].columnOid];
std::cerr << "string [" << dictStrList[i][j].safeString("<<null>>") << "]" << std::endl;
bool isToken = colStructList[tokenOidIdx].colType == WriteEngine::WR_TOKEN &&
colStructList[tokenOidIdx].tokenFlag;
if (isToken && !colValueList[tokenOidIdx][j].data.empty())
{
Token t = boost::any_cast<Token>(colValueList[tokenOidIdx][j].data);
std::cerr << "Token: block pos:[" << t.op << "] fbo: [" << t.fbo << "] bc: [" << t.bc << "]"
<< std::endl;
}
}
}
}
std::cerr << "=========================" << std::endl;
}
/***********************************************************
* DESCRIPTION:
* Process version buffer before any write operation
* PARAMETERS:
* txnid - transaction id
* oid - column oid
* totalRow - total number of rows
* rowIdArray - rowid array
* RETURN:
* NO_ERROR if success
* others if something wrong in inserting the value
***********************************************************/
int WriteEngineWrapper::processVersionBuffer(IDBDataFile* pFile, const TxnID& txnid,
const ColStruct& colStruct, int width, int totalRow,
const RID* rowIdArray, vector<LBIDRange>& rangeList)
{
if (idbdatafile::IDBPolicy::useHdfs())
return 0;
RID curRowId;
int rc = NO_ERROR;
int curFbo = 0, curBio, lastFbo = -1;
bool successFlag;
BRM::LBID_t lbid;
BRM::VER_t verId = (BRM::VER_t)txnid;
vector<uint32_t> fboList;
LBIDRange range;
ColumnOp* colOp = m_colOp[op(colStruct.fCompressionType)];
for (int i = 0; i < totalRow; i++)
{
curRowId = rowIdArray[i];
// cout << "processVersionBuffer got rid " << curRowId << endl;
successFlag = colOp->calculateRowId(curRowId, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
// cout << "processVersionBuffer is processing lbid " << lbid << endl;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getBrmInfo(colStruct.dataOid, colStruct.fColPartition,
colStruct.fColSegment, curFbo, lbid));
// cout << "processVersionBuffer is processing lbid " << lbid << endl;
fboList.push_back((uint32_t)curFbo);
range.start = lbid;
range.size = 1;
rangeList.push_back(range);
}
lastFbo = curFbo;
}
}
std::vector<VBRange> freeList;
rc = BRMWrapper::getInstance()->writeVB(pFile, verId, colStruct.dataOid, fboList, rangeList, colOp,
freeList, colStruct.fColDbRoot);
return rc;
}
int WriteEngineWrapper::processVersionBuffers(IDBDataFile* pFile, const TxnID& txnid,
const ColStruct& colStruct, int width, int totalRow,
const RIDList& ridList, vector<LBIDRange>& rangeList)
{
if (idbdatafile::IDBPolicy::useHdfs())
return 0;
RID curRowId;
int rc = NO_ERROR;
int curFbo = 0, curBio, lastFbo = -1;
bool successFlag;
BRM::LBID_t lbid;
BRM::VER_t verId = (BRM::VER_t)txnid;
LBIDRange range;
vector<uint32_t> fboList;
// vector<LBIDRange> rangeList;
ColumnOp* colOp = m_colOp[op(colStruct.fCompressionType)];
for (int i = 0; i < totalRow; i++)
{
curRowId = ridList[i];
// cout << "processVersionBuffer got rid " << curRowId << endl;
successFlag = colOp->calculateRowId(curRowId, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
// cout << "processVersionBuffer is processing lbid " << lbid << endl;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getBrmInfo(colStruct.dataOid, colStruct.fColPartition,
colStruct.fColSegment, curFbo, lbid));
// cout << "processVersionBuffer is processing lbid " << lbid << endl;
fboList.push_back((uint32_t)curFbo);
range.start = lbid;
range.size = 1;
rangeList.push_back(range);
}
lastFbo = curFbo;
}
}
// cout << "calling writeVB with blocks " << rangeList.size() << endl;
std::vector<VBRange> freeList;
rc = BRMWrapper::getInstance()->writeVB(pFile, verId, colStruct.dataOid, fboList, rangeList, colOp,
freeList, colStruct.fColDbRoot);
return rc;
}
int WriteEngineWrapper::processBeginVBCopy(const TxnID& txnid, const vector<ColStruct>& colStructList,
const RIDList& ridList, std::vector<VBRange>& freeList,
vector<vector<uint32_t> >& fboLists,
vector<vector<LBIDRange> >& rangeLists,
vector<LBIDRange>& rangeListTot)
{
if (idbdatafile::IDBPolicy::useHdfs())
return 0;
RID curRowId;
int rc = NO_ERROR;
int curFbo = 0, curBio, lastFbo = -1;
bool successFlag;
BRM::LBID_t lbid;
LBIDRange range;
// StopWatch timer;
// timer.start("calculation");
for (uint32_t j = 0; j < colStructList.size(); j++)
{
vector<uint32_t> fboList;
vector<LBIDRange> rangeList;
lastFbo = -1;
ColumnOp* colOp = m_colOp[op(colStructList[j].fCompressionType)];
ColStruct curColStruct = colStructList[j];
Convertor::convertColType(&curColStruct);
for (uint32_t i = 0; i < ridList.size(); i++)
{
curRowId = ridList[i];
// cout << "processVersionBuffer got rid " << curRowId << endl;
successFlag = colOp->calculateRowId(curRowId, BYTE_PER_BLOCK / curColStruct.colWidth,
curColStruct.colWidth, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
// cout << "processVersionBuffer is processing curFbo " << curFbo << endl;
RETURN_ON_ERROR(BRMWrapper::getInstance()->getBrmInfo(colStructList[j].dataOid,
colStructList[j].fColPartition,
colStructList[j].fColSegment, curFbo, lbid));
// cout << "beginVBCopy is processing lbid:transaction " << lbid <<":"<<txnid<< endl;
fboList.push_back((uint32_t)curFbo);
range.start = lbid;
range.size = 1;
rangeList.push_back(range);
}
lastFbo = curFbo;
}
}
BRMWrapper::getInstance()->pruneLBIDList(txnid, &rangeList, &fboList);
rangeLists.push_back(rangeList);
fboLists.push_back(fboList);
rangeListTot.insert(rangeListTot.end(), rangeList.begin(), rangeList.end());
}
if (rangeListTot.size() > 0)
rc = BRMWrapper::getInstance()->getDbrmObject()->beginVBCopy(txnid, colStructList[0].fColDbRoot,
rangeListTot, freeList);
// timer.stop("beginVBCopy");
// timer.finish();
return rc;
}
/**
* @brief Process versioning for batch insert - only version the hwm block.
*/
#if 0
int WriteEngineWrapper::processBatchVersions(const TxnID& txnid, std::vector<Column> columns, std::vector<BRM::LBIDRange>& rangeList)
{
int rc = 0;
std::vector<DbFileOp*> fileOps;
//open the column files
for ( unsigned i = 0; i < columns.size(); i++)
{
ColumnOp* colOp = m_colOp[op(columns[i].compressionType)];
Column curCol;
// set params
colOp->initColumn(curCol);
ColType colType;
Convertor::convertColType(columns[i].colDataType, colType);
colOp->setColParam(curCol, 0, columns[i].colWidth,
columns[i].colDataType, colType, columns[i].dataFile.oid,
columns[i].compressionType,
columns[i].dataFile.fDbRoot, columns[i].dataFile.fPartition, columns[i].dataFile.fSegment);
string segFile;
rc = colOp->openColumnFile(curCol, segFile, IO_BUFF_SIZE);
if (rc != NO_ERROR)
break;
columns[i].dataFile.pFile = curCol.dataFile.pFile;
fileOps.push_back(colOp);
}
if ( rc == 0)
{
BRM::VER_t verId = (BRM::VER_t) txnid;
rc = BRMWrapper::getInstance()->writeBatchVBs(verId, columns, rangeList, fileOps);
}
//close files
for ( unsigned i = 0; i < columns.size(); i++)
{
ColumnOp* colOp = dynamic_cast<ColumnOp*> (fileOps[i]);
Column curCol;
// set params
colOp->initColumn(curCol);
ColType colType;
Convertor::convertColType(columns[i].colDataType, colType);
colOp->setColParam(curCol, 0, columns[i].colWidth,
columns[i].colDataType, colType, columns[i].dataFile.oid,
columns[i].compressionType,
columns[i].dataFile.fDbRoot, columns[i].dataFile.fPartition, columns[i].dataFile.fSegment);
curCol.dataFile.pFile = columns[i].dataFile.pFile;
colOp->clearColumn(curCol);
}
return rc;
}
#endif
void WriteEngineWrapper::writeVBEnd(const TxnID& txnid, std::vector<BRM::LBIDRange>& rangeList)
{
if (idbdatafile::IDBPolicy::useHdfs())
return;
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
}
int WriteEngineWrapper::updateColumnRec(const TxnID& txnid, const vector<CSCTypesList>& colExtentsColType,
vector<ColStructList>& colExtentsStruct, ColValueList& colValueList,
vector<void*>& colOldValueList, vector<RIDList>& ridLists,
vector<DctnryStructList>& dctnryExtentsStruct,
DctnryValueList& dctnryValueList, const int32_t tableOid,
bool hasAUXCol)
{
int rc = 0;
unsigned numExtents = colExtentsStruct.size();
ColStructList colStructList;
DctnryStructList dctnryStructList;
WriteEngine::CSCTypesList cscColTypeList;
ColumnOp* colOp = NULL;
ExtCPInfoList infosToUpdate;
for (unsigned extent = 0; extent < numExtents; extent++)
{
colStructList = colExtentsStruct[extent];
dctnryStructList = dctnryExtentsStruct[extent];
cscColTypeList = colExtentsColType[extent];
if (m_opType != DELETE)
{
// Tokenize data if needed
vector<Token> tokenList;
DctColTupleList::iterator dctCol_iter;
ColTupleList::iterator col_iter;
for (unsigned i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
// only need to tokenize once
dctCol_iter = dctnryValueList[i].begin();
Token token;
if (!dctCol_iter->isNull)
{
RETURN_ON_ERROR_REPORT(
tokenize(txnid, dctnryStructList[i], *dctCol_iter, true)); // @bug 5572 HDFS tmp file
token = dctCol_iter->token;
#ifdef PROFILE
// timer.stop("tokenize");
#endif
}
tokenList.push_back(token);
}
}
int dicPos = 0;
for (unsigned i = 0; i < colStructList.size(); i++)
{
if (colStructList[i].tokenFlag)
{
// only need to tokenize once
col_iter = colValueList[i].begin();
while (col_iter != colValueList[i].end())
{
col_iter->data = tokenList[dicPos];
col_iter++;
}
dicPos++;
}
}
}
RIDList::iterator rid_iter;
// Mark extents invalid
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
rid_iter = ridLists[extent].begin();
RID aRid = *rid_iter;
ExtCPInfoList currentExtentRanges;
for (const auto& colStruct : colStructList)
{
currentExtentRanges.push_back(
ExtCPInfo(colStruct.colDataType, colStruct.colWidth)); // temporary for each extent.
}
std::vector<ExtCPInfo*> currentExtentRangesPtrs(colStructList.size(), NULL); // pointers for each extent.
if (m_opType != DELETE)
m_opType = UPDATE;
for (unsigned j = 0; j < colStructList.size(); j++)
{
colOp = m_colOp[op(colStructList[j].fCompressionType)];
ExtCPInfo* cpInfoP = &(currentExtentRanges[j]);
cpInfoP = getCPInfoToUpdateForUpdatableType(colStructList[j], cpInfoP, m_opType);
currentExtentRangesPtrs[j] = cpInfoP;
// XXX: highly dubious.
// if (!colStructList[j].tokenFlag)
// continue;
width = colOp->getCorrectRowWidth(colStructList[j].colDataType, colStructList[j].colWidth);
successFlag = colOp->calculateRowId(aRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR_REPORT(AddLBIDtoList(txnid, colStructList[j], curFbo, cpInfoP));
}
}
}
//#ifdef PROFILE
// timer.start("markExtentsInvalid");
//#endif
bool hasFastDelete = false;
if (m_opType == DELETE && hasAUXCol)
{
hasFastDelete = Config::getFastDelete();
}
if (hasFastDelete)
{
ColStructList colStructListAUX(1, colStructList.back());
WriteEngine::CSCTypesList cscColTypeListAUX(1, cscColTypeList.back());
ColValueList colValueListAUX(1, colValueList.back());
std::vector<ExtCPInfo*> currentExtentRangesPtrsAUX(1, currentExtentRangesPtrs.back());
rc = writeColumnRecUpdate(txnid, cscColTypeListAUX, colStructListAUX, colValueListAUX, colOldValueList,
ridLists[extent], tableOid, true, ridLists[extent].size(),
&currentExtentRangesPtrsAUX, hasAUXCol);
for (auto& cpInfoPtr : currentExtentRangesPtrs)
{
if (cpInfoPtr)
{
cpInfoPtr->toInvalid();
}
}
}
else
{
rc = writeColumnRecUpdate(txnid, cscColTypeList, colStructList, colValueList, colOldValueList,
ridLists[extent], tableOid, true, ridLists[extent].size(),
&currentExtentRangesPtrs, hasAUXCol);
}
if (rc != NO_ERROR)
break;
// copy updated ranges into bulk update vector.
for (auto cpInfoPtr : currentExtentRangesPtrs)
{
if (cpInfoPtr)
{
cpInfoPtr->fCPInfo.seqNum++;
infosToUpdate.push_back(*cpInfoPtr);
}
}
}
markTxnExtentsAsInvalid(txnid);
if (rc == NO_ERROR)
{
ExtCPInfoList infosToDrop = infosToUpdate;
for (auto& cpInfo : infosToDrop)
{
cpInfo.fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
}
// ZZZZ
rc = BRMWrapper::getInstance()->setExtentsMaxMin(infosToDrop);
setInvalidCPInfosSpecialMarks(infosToUpdate);
rc = BRMWrapper::getInstance()->setExtentsMaxMin(infosToUpdate);
}
return rc;
}
int WriteEngineWrapper::updateColumnRecs(const TxnID& txnid, const CSCTypesList& cscColTypeList,
vector<ColStruct>& colExtentsStruct, ColValueList& colValueList,
const RIDList& ridLists, const int32_t tableOid)
{
// Mark extents invalid
ColumnOp* colOp = NULL;
bool successFlag = true;
unsigned width = 0;
int curFbo = 0, curBio, lastFbo = -1;
RID aRid = ridLists[0];
int rc = 0;
ExtCPInfoList infosToUpdate;
for (const auto& colStruct : colExtentsStruct)
{
infosToUpdate.push_back(ExtCPInfo(colStruct.colDataType, colStruct.colWidth));
}
ExtCPInfoList bulkUpdateInfos;
std::vector<ExtCPInfo*> pointersToInfos; // pointersToInfos[i] points to infosToUpdate[i] and may be NULL.
m_opType = UPDATE;
for (unsigned j = 0; j < colExtentsStruct.size(); j++)
{
colOp = m_colOp[op(colExtentsStruct[j].fCompressionType)];
ExtCPInfo* cpInfoP = &(infosToUpdate[j]);
cpInfoP = getCPInfoToUpdateForUpdatableType(colExtentsStruct[j], cpInfoP, m_opType);
pointersToInfos.push_back(cpInfoP);
width = colOp->getCorrectRowWidth(colExtentsStruct[j].colDataType, colExtentsStruct[j].colWidth);
successFlag = colOp->calculateRowId(aRid, BYTE_PER_BLOCK / width, width, curFbo, curBio);
if (successFlag)
{
if (curFbo != lastFbo)
{
RETURN_ON_ERROR(AddLBIDtoList(txnid, colExtentsStruct[j], curFbo, cpInfoP));
}
}
}
markTxnExtentsAsInvalid(txnid);
if (m_opType != DELETE)
m_opType = UPDATE;
for (auto cpInfoP : pointersToInfos)
{
if (cpInfoP)
{
auto tmp = *cpInfoP;
tmp.fCPInfo.seqNum = SEQNUM_MARK_INVALID_SET_RANGE;
bulkUpdateInfos.push_back(tmp);
}
}
if (!bulkUpdateInfos.empty())
{
rc = BRMWrapper::getInstance()->setExtentsMaxMin(bulkUpdateInfos);
}
rc = writeColumnRecords(txnid, cscColTypeList, colExtentsStruct, colValueList, ridLists, tableOid, true,
&pointersToInfos);
if (rc == NO_ERROR)
{
bulkUpdateInfos.clear();
for (auto cpInfoP : pointersToInfos)
{
if (cpInfoP)
{
cpInfoP->fCPInfo.seqNum++;
bulkUpdateInfos.push_back(*cpInfoP);
}
}
if (!bulkUpdateInfos.empty())
{
setInvalidCPInfosSpecialMarks(bulkUpdateInfos);
rc = BRMWrapper::getInstance()->setExtentsMaxMin(bulkUpdateInfos);
}
}
m_opType = NOOP;
return rc;
}
int WriteEngineWrapper::writeColumnRecords(const TxnID& txnid, const CSCTypesList& cscColTypeList,
vector<ColStruct>& colStructList, ColValueList& colValueList,
const RIDList& ridLists, const int32_t tableOid, bool versioning,
std::vector<ExtCPInfo*>* cpInfos)
{
bool bExcp;
int rc = 0;
void* valArray = NULL;
void* oldValArray = NULL;
Column curCol;
ColStruct curColStruct;
CalpontSystemCatalog::ColType curColType;
ColTupleList curTupleList;
ColStructList::size_type totalColumn;
ColStructList::size_type i;
ColTupleList::size_type totalRow;
setTransId(txnid);
totalColumn = colStructList.size();
totalRow = ridLists.size();
TableMetaData* aTbaleMetaData = TableMetaData::makeTableMetaData(tableOid);
for (i = 0; i < totalColumn; i++)
{
ExtCPInfo* cpInfo = NULL;
if (cpInfos)
{
cpInfo = (*cpInfos)[i];
}
valArray = NULL;
oldValArray = NULL;
curColStruct = colStructList[i];
curColType = cscColTypeList[i];
curTupleList = colValueList[i];
ColumnOp* colOp = m_colOp[op(curColStruct.fCompressionType)];
Convertor::convertColType(&curColStruct);
// set params
colOp->initColumn(curCol);
colOp->setColParam(curCol, 0, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, curColStruct.fColDbRoot,
curColStruct.fColPartition, curColStruct.fColSegment);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(curColStruct.dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == curColStruct.fColDbRoot) && (it->partNum == curColStruct.fColPartition) &&
(it->segNum == curColStruct.fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = curColStruct.fColDbRoot;
aExt.partNum = curColStruct.fColPartition;
aExt.segNum = curColStruct.fColSegment;
aExt.compType = curColStruct.fCompressionType;
aExt.isDict = false;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
string segFile;
rc = colOp->openColumnFile(curCol, segFile, true); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffers(curCol.dataFile.pFile, txnid, curColStruct, curColStruct.colWidth, totalRow,
ridLists, rangeList);
}
if (rc != NO_ERROR)
{
if (curColStruct.fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
allocateValArray(valArray, totalRow, curColStruct.colType, curColStruct.colWidth);
if (m_opType != INSERT && cpInfo)
{
allocateValArray(oldValArray, totalRow, curColStruct.colType, curColStruct.colWidth);
}
// convert values to valArray
bExcp = false;
try
{
convertValArray(totalRow, cscColTypeList[i], curColStruct.colType, curTupleList, valArray);
}
catch (...)
{
bExcp = true;
}
if (bExcp)
{
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
return ERR_PARSING;
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRowsValues(curCol, totalRow, ridLists, valArray, oldValArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
colOp->clearColumn(curCol);
updateMaxMinRange(totalRow, totalRow, cscColTypeList[i], curColStruct.colType, valArray, oldValArray,
cpInfo, false);
if (curColStruct.fCompressionType == 0)
{
std::vector<BRM::FileInfo> files;
BRM::FileInfo aFile;
aFile.partitionNum = curColStruct.fColPartition;
aFile.dbRoot = curColStruct.fColDbRoot;
;
aFile.segmentNum = curColStruct.fColSegment;
aFile.compType = curColStruct.fCompressionType;
files.push_back(aFile);
if (idbdatafile::IDBPolicy::useHdfs())
cacheutils::purgePrimProcFdCache(files, Config::getLocalModuleID());
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
if (valArray != NULL)
free(valArray);
if (oldValArray != NULL)
{
free(oldValArray);
}
// check error
if (rc != NO_ERROR)
break;
}
return rc;
}
/*@brief writeColumnRec - Write values to a column
*/
/***********************************************************
* DESCRIPTION:
* Write values to a column
* PARAMETERS:
* tableOid - table object id
* cscColTypesList - CSC ColType list
* colStructList - column struct list
* colValueList - column value list
* colNewStructList - the new extent struct list
* colNewValueList - column value list for the new extent
* rowIdArray - row id list
* useTmpSuffix - use temp suffix for db output file
* RETURN:
* NO_ERROR if success
* others if something wrong in inserting the value
***********************************************************/
int WriteEngineWrapper::writeColumnRec(const TxnID& txnid, const CSCTypesList& cscColTypeList,
const ColStructList& colStructList, ColValueList& colValueList,
RID* rowIdArray, const ColStructList& newColStructList,
ColValueList& newColValueList, const int32_t tableOid,
bool useTmpSuffix, bool versioning, ColSplitMaxMinInfoList* maxMins)
{
bool bExcp;
int rc = 0;
void* valArray;
void* oldValArray;
string segFile;
Column curCol;
ColTupleList oldTupleList;
ColStructList::size_type totalColumn;
ColStructList::size_type i;
ColTupleList::size_type totalRow1, totalRow2;
setTransId(txnid);
totalColumn = colStructList.size();
#ifdef PROFILE
StopWatch timer;
#endif
totalRow1 = colValueList[0].size();
totalRow2 = 0;
if (newColValueList.size() > 0)
totalRow2 = newColValueList[0].size();
TableMetaData* aTbaleMetaData = TableMetaData::makeTableMetaData(tableOid);
for (i = 0; i < totalColumn; i++)
{
if (totalRow2 > 0)
{
RID* secondPart = rowIdArray + totalRow1;
//@Bug 2205 Check if all rows go to the new extent
if (totalRow1 > 0)
{
// Write the first batch
valArray = NULL;
oldValArray = NULL;
RID* firstPart = rowIdArray;
ColumnOp* colOp = m_colOp[op(colStructList[i].fCompressionType)];
// set params
colOp->initColumn(curCol);
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, 0, colStructList[i].colWidth, colStructList[i].colDataType,
colStructList[i].colType, colStructList[i].dataOid,
colStructList[i].fCompressionType, colStructList[i].fColDbRoot,
colStructList[i].fColPartition, colStructList[i].fColSegment);
colOp->findTypeHandler(colStructList[i].colWidth, colStructList[i].colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(colStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) &&
(it->partNum == colStructList[i].fColPartition) && (it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = colStructList[i].fColDbRoot;
aExt.partNum = colStructList[i].fColPartition;
aExt.segNum = colStructList[i].fColSegment;
aExt.compType = colStructList[i].fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
// handling versioning
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffer(curCol.dataFile.pFile, txnid, colStructList[i], colStructList[i].colWidth,
totalRow1, firstPart, rangeList);
if (rc != NO_ERROR)
{
if (colStructList[i].fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
}
// WIP We can allocate based on column size and not colType
// have to init the size here
allocateValArray(valArray, totalRow1, colStructList[i].colType, colStructList[i].colWidth);
ExtCPInfo* cpInfo = getCPInfoToUpdateForUpdatableType(
colStructList[i], maxMins ? ((*maxMins)[i]).fSplitMaxMinInfoPtrs[0] : NULL, m_opType);
if (m_opType != INSERT && cpInfo != NULL) // we allocate space for old values only when we need them.
{
allocateValArray(oldValArray, totalRow1, colStructList[i].colType, colStructList[i].colWidth);
}
else
{
oldValArray = NULL;
}
// convert values to valArray
// WIP Is m_opType ever set to DELETE?
if (m_opType != DELETE)
{
bExcp = false;
try
{
// WIP We convert values twice!?
// dmlcommandproc converts strings to boost::any and this converts
// into actual type value masked by *void
// It is not clear why we need to convert to boost::any b/c we can convert from the original
// string here
convertValArray(totalRow1, cscColTypeList[i], colStructList[i].colType, colValueList[i],
valArray);
}
catch (...)
{
bExcp = true;
}
if (bExcp)
{
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
return ERR_PARSING;
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow1, firstPart, valArray, oldValArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
else
{
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow1, rowIdArray, valArray, oldValArray, true);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
colOp->clearColumn(curCol);
updateMaxMinRange(totalRow1, totalRow1, cscColTypeList[i], colStructList[i].colType, valArray,
oldValArray, cpInfo, rowIdArray[0] == 0 && m_opType == INSERT);
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
if (valArray != NULL)
free(valArray);
if (oldValArray != NULL)
free(oldValArray);
// check error
if (rc != NO_ERROR)
break;
}
// Process the second batch
valArray = NULL;
oldValArray = NULL;
ColumnOp* colOp = m_colOp[op(newColStructList[i].fCompressionType)];
// set params
colOp->initColumn(curCol);
colOp->setColParam(curCol, 0, newColStructList[i].colWidth, newColStructList[i].colDataType,
newColStructList[i].colType, newColStructList[i].dataOid,
newColStructList[i].fCompressionType, newColStructList[i].fColDbRoot,
newColStructList[i].fColPartition, newColStructList[i].fColSegment);
colOp->findTypeHandler(newColStructList[i].colWidth, newColStructList[i].colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(newColStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newColStructList[i].fColDbRoot) &&
(it->partNum == newColStructList[i].fColPartition) &&
(it->segNum == newColStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = newColStructList[i].fColDbRoot;
aExt.partNum = newColStructList[i].fColPartition;
aExt.segNum = newColStructList[i].fColSegment;
aExt.compType = newColStructList[i].fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(newColStructList[i].dataOid, aColExtsInfo);
}
// Pass "false" for hdfs tmp file flag. Since we only allow 1
// extent per segment file (with HDFS), we can assume a second
// extent is going to a new file (and won't need tmp file).
rc = colOp->openColumnFile(curCol, segFile, false, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
// handling versioning
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffer(curCol.dataFile.pFile, txnid, newColStructList[i],
newColStructList[i].colWidth, totalRow2, secondPart, rangeList);
if (rc != NO_ERROR)
{
if (newColStructList[i].fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
}
ExtCPInfo* cpInfo = getCPInfoToUpdateForUpdatableType(
newColStructList[i], maxMins ? ((*maxMins)[i]).fSplitMaxMinInfoPtrs[1] : NULL, m_opType);
allocateValArray(valArray, totalRow2, newColStructList[i].colType, newColStructList[i].colWidth);
if (m_opType != INSERT && cpInfo != NULL) // we allocate space for old values only when we need them.
{
allocateValArray(oldValArray, totalRow1, colStructList[i].colType, colStructList[i].colWidth);
}
else
{
oldValArray = NULL;
}
// convert values to valArray
if (m_opType != DELETE)
{
bExcp = false;
try
{
convertValArray(totalRow2, cscColTypeList[i], newColStructList[i].colType, newColValueList[i],
valArray);
}
catch (...)
{
bExcp = true;
}
if (bExcp)
{
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
return ERR_PARSING;
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(
curCol, totalRow2, secondPart, valArray,
oldValArray); // XXX: here we use secondPart array and just below we use rowIdArray. WHY???
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
else
{
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(
curCol, totalRow2, rowIdArray, valArray, oldValArray,
true); // XXX: BUG: here we use rowIdArray and just above we use secondPart array. WHY???
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
colOp->clearColumn(curCol);
updateMaxMinRange(totalRow2, totalRow2, cscColTypeList[i], newColStructList[i].colType, valArray,
oldValArray, cpInfo, secondPart[0] == 0 && m_opType == INSERT);
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
if (valArray != NULL)
free(valArray);
// check error
if (rc != NO_ERROR)
break;
}
else
{
valArray = NULL;
oldValArray = NULL;
ColumnOp* colOp = m_colOp[op(colStructList[i].fCompressionType)];
ExtCPInfo* cpInfo = getCPInfoToUpdateForUpdatableType(
colStructList[i], maxMins ? ((*maxMins)[i]).fSplitMaxMinInfoPtrs[0] : NULL, m_opType);
// set params
colOp->initColumn(curCol);
colOp->setColParam(curCol, 0, colStructList[i].colWidth, colStructList[i].colDataType,
colStructList[i].colType, colStructList[i].dataOid,
colStructList[i].fCompressionType, colStructList[i].fColDbRoot,
colStructList[i].fColPartition, colStructList[i].fColSegment);
colOp->findTypeHandler(colStructList[i].colWidth, colStructList[i].colDataType);
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
// cout << " Opened file oid " << curCol.dataFile.pFile << endl;
if (rc != NO_ERROR)
break;
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(colStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = colStructList[i].fColDbRoot;
aExt.partNum = colStructList[i].fColPartition;
aExt.segNum = colStructList[i].fColSegment;
aExt.compType = colStructList[i].fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
// handling versioning
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffer(curCol.dataFile.pFile, txnid, colStructList[i], colStructList[i].colWidth,
totalRow1, rowIdArray, rangeList);
if (rc != NO_ERROR)
{
if (colStructList[i].fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
}
allocateValArray(valArray, totalRow1, colStructList[i].colType, colStructList[i].colWidth);
if (m_opType != INSERT && cpInfo != NULL) // we allocate space for old values only when we need them.
{
allocateValArray(oldValArray, totalRow1, colStructList[i].colType, colStructList[i].colWidth);
}
else
{
oldValArray = NULL;
}
// convert values to valArray
if (m_opType != DELETE)
{
bExcp = false;
try
{
convertValArray(totalRow1, cscColTypeList[i], colStructList[i].colType, colValueList[i], valArray,
true);
}
catch (...)
{
bExcp = true;
}
if (bExcp)
{
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
return ERR_PARSING;
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow1, rowIdArray, valArray, oldValArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
else
{
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow1, rowIdArray, valArray, oldValArray, true);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
colOp->clearColumn(curCol);
updateMaxMinRange(totalRow1, totalRow1, cscColTypeList[i], colStructList[i].colType, valArray,
oldValArray, cpInfo, rowIdArray[0] == 0 && m_opType == INSERT);
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
if (valArray != NULL)
free(valArray);
if (oldValArray != NULL)
free(oldValArray);
// check error
if (rc != NO_ERROR)
break;
}
} // end of for (i = 0
#ifdef PROFILE
timer.finish();
#endif
return rc;
}
int WriteEngineWrapper::writeColumnRecBinary(const TxnID& txnid, const ColStructList& colStructList,
std::vector<uint64_t>& colValueList, RID* rowIdArray,
const ColStructList& newColStructList,
std::vector<uint64_t>& newColValueList, const int32_t tableOid,
bool useTmpSuffix, bool versioning)
{
int rc = 0;
void* valArray = NULL;
string segFile;
Column curCol;
ColStructList::size_type totalColumn;
ColStructList::size_type i;
size_t totalRow1, totalRow2;
setTransId(txnid);
totalColumn = colStructList.size();
#ifdef PROFILE
StopWatch timer;
#endif
totalRow1 = colValueList.size() / totalColumn;
if (newColValueList.size() > 0)
{
totalRow2 = newColValueList.size() / newColStructList.size();
totalRow1 -= totalRow2;
}
else
{
totalRow2 = 0;
}
// It is possible totalRow1 is zero but totalRow2 has values
if ((totalRow1 == 0) && (totalRow2 == 0))
return rc;
TableMetaData* aTbaleMetaData = TableMetaData::makeTableMetaData(tableOid);
if (totalRow1)
{
valArray = malloc(sizeof(uint64_t) * totalRow1);
for (i = 0; i < totalColumn; i++)
{
//@Bug 2205 Check if all rows go to the new extent
// Write the first batch
RID* firstPart = rowIdArray;
ColumnOp* colOp = m_colOp[op(colStructList[i].fCompressionType)];
// set params
colOp->initColumn(curCol);
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, 0, colStructList[i].colWidth, colStructList[i].colDataType,
colStructList[i].colType, colStructList[i].dataOid,
colStructList[i].fCompressionType, colStructList[i].fColDbRoot,
colStructList[i].fColPartition, colStructList[i].fColSegment);
colOp->findTypeHandler(colStructList[i].colWidth, colStructList[i].colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(colStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == colStructList[i].fColDbRoot) && (it->partNum == colStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = colStructList[i].fColDbRoot;
aExt.partNum = colStructList[i].fColPartition;
aExt.segNum = colStructList[i].fColSegment;
aExt.compType = colStructList[i].fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
// handling versioning
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffer(curCol.dataFile.pFile, txnid, colStructList[i], colStructList[i].colWidth,
totalRow1, firstPart, rangeList);
if (rc != NO_ERROR)
{
if (colStructList[i].fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
}
// totalRow1 -= totalRow2;
// have to init the size here
// nullArray = (bool*) malloc(sizeof(bool) * totalRow);
uint8_t tmp8;
uint16_t tmp16;
uint32_t tmp32;
for (size_t j = 0; j < totalRow1; j++)
{
uint64_t curValue = colValueList[((totalRow1 + totalRow2) * i) + j];
switch (colStructList[i].colType)
{
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT: ((uint64_t*)valArray)[j] = curValue; break;
case WriteEngine::WR_INT:
case WriteEngine::WR_UINT:
case WriteEngine::WR_MEDINT:
case WriteEngine::WR_UMEDINT:
case WriteEngine::WR_FLOAT:
tmp32 = curValue;
((uint32_t*)valArray)[j] = tmp32;
break;
case WriteEngine::WR_ULONGLONG:
case WriteEngine::WR_LONGLONG:
case WriteEngine::WR_DOUBLE:
case WriteEngine::WR_TOKEN: ((uint64_t*)valArray)[j] = curValue; break;
case WriteEngine::WR_BYTE:
case WriteEngine::WR_UBYTE:
tmp8 = curValue;
((uint8_t*)valArray)[j] = tmp8;
break;
case WriteEngine::WR_SHORT:
case WriteEngine::WR_USHORT:
tmp16 = curValue;
((uint16_t*)valArray)[j] = tmp16;
break;
case WriteEngine::WR_BINARY: ((uint64_t*)valArray)[j] = curValue; break;
}
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow1, firstPart, valArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
colOp->closeColumnFile(curCol);
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
// check error
if (rc != NO_ERROR)
break;
} // end of for (i = 0
if (valArray != NULL)
{
free(valArray);
valArray = NULL;
}
}
// MCOL-1176 - Write second extent
if (totalRow2)
{
valArray = malloc(sizeof(uint64_t) * totalRow2);
for (i = 0; i < newColStructList.size(); i++)
{
//@Bug 2205 Check if all rows go to the new extent
// Write the first batch
RID* secondPart = rowIdArray + totalRow1;
ColumnOp* colOp = m_colOp[op(newColStructList[i].fCompressionType)];
// set params
colOp->initColumn(curCol);
// need to pass real dbRoot, partition, and segment to setColParam
colOp->setColParam(curCol, 0, newColStructList[i].colWidth, newColStructList[i].colDataType,
newColStructList[i].colType, newColStructList[i].dataOid,
newColStructList[i].fCompressionType, newColStructList[i].fColDbRoot,
newColStructList[i].fColPartition, newColStructList[i].fColSegment);
colOp->findTypeHandler(newColStructList[i].colWidth, newColStructList[i].colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(newColStructList[i].dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == newColStructList[i].fColDbRoot) &&
(it->partNum == newColStructList[i].fColPartition) &&
(it->segNum == colStructList[i].fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = newColStructList[i].fColDbRoot;
aExt.partNum = newColStructList[i].fColPartition;
aExt.segNum = newColStructList[i].fColSegment;
aExt.compType = newColStructList[i].fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(newColStructList[i].dataOid, aColExtsInfo);
}
rc = colOp->openColumnFile(curCol, segFile, useTmpSuffix, IO_BUFF_SIZE); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
// handling versioning
vector<LBIDRange> rangeList;
if (versioning)
{
rc = processVersionBuffer(curCol.dataFile.pFile, txnid, newColStructList[i],
newColStructList[i].colWidth, totalRow2, secondPart, rangeList);
if (rc != NO_ERROR)
{
if (newColStructList[i].fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
break;
}
}
// totalRow1 -= totalRow2;
// have to init the size here
// nullArray = (bool*) malloc(sizeof(bool) * totalRow);
uint8_t tmp8;
uint16_t tmp16;
uint32_t tmp32;
for (size_t j = 0; j < totalRow2; j++)
{
uint64_t curValue = newColValueList[(totalRow2 * i) + j];
switch (newColStructList[i].colType)
{
case WriteEngine::WR_VARBINARY: // treat same as char for now
case WriteEngine::WR_CHAR:
case WriteEngine::WR_BLOB:
case WriteEngine::WR_TEXT: ((uint64_t*)valArray)[j] = curValue; break;
case WriteEngine::WR_INT:
case WriteEngine::WR_UINT:
case WriteEngine::WR_MEDINT:
case WriteEngine::WR_UMEDINT:
case WriteEngine::WR_FLOAT:
tmp32 = curValue;
((uint32_t*)valArray)[j] = tmp32;
break;
case WriteEngine::WR_ULONGLONG:
case WriteEngine::WR_LONGLONG:
case WriteEngine::WR_DOUBLE:
case WriteEngine::WR_TOKEN: ((uint64_t*)valArray)[j] = curValue; break;
case WriteEngine::WR_BYTE:
case WriteEngine::WR_UBYTE:
tmp8 = curValue;
((uint8_t*)valArray)[j] = tmp8;
break;
case WriteEngine::WR_SHORT:
case WriteEngine::WR_USHORT:
tmp16 = curValue;
((uint16_t*)valArray)[j] = tmp16;
break;
case WriteEngine::WR_BINARY: ((uint64_t*)valArray)[j] = curValue; break;
}
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRow(curCol, totalRow2, secondPart, valArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
colOp->closeColumnFile(curCol);
if (versioning)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeList);
// check error
if (rc != NO_ERROR)
break;
} // end of for (i = 0
}
if (valArray != NULL)
free(valArray);
#ifdef PROFILE
timer.finish();
#endif
return rc;
}
int WriteEngineWrapper::writeColumnRecUpdate(const TxnID& txnid, const CSCTypesList& cscColTypeList,
const ColStructList& colStructList,
const ColValueList& colValueList, vector<void*>& colOldValueList,
const RIDList& ridList, const int32_t tableOid,
bool convertStructFlag, ColTupleList::size_type nRows,
std::vector<ExtCPInfo*>* cpInfos,
bool hasAUXCol)
{
bool bExcp;
int rc = 0;
void* valArray = NULL;
void* oldValArray = NULL;
Column curCol;
ColStruct curColStruct;
ColTupleList curTupleList, oldTupleList;
ColStructList::size_type totalColumn;
ColStructList::size_type i;
ColTupleList::size_type totalRow;
setTransId(txnid);
colOldValueList.clear();
totalColumn = colStructList.size();
totalRow = nRows;
#ifdef PROFILE
StopWatch timer;
#endif
vector<LBIDRange> rangeListTot;
std::vector<VBRange> freeList;
vector<vector<uint32_t> > fboLists;
vector<vector<LBIDRange> > rangeLists;
rc = processBeginVBCopy(txnid, ((m_opType == DELETE && hasAUXCol) ? ColStructList(1, colStructList.back()) : colStructList),
ridList, freeList, fboLists, rangeLists, rangeListTot);
if (rc != NO_ERROR)
{
if (rangeListTot.size() > 0)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeListTot);
switch (rc)
{
case BRM::ERR_DEADLOCK: return ERR_BRM_DEAD_LOCK;
case BRM::ERR_VBBM_OVERFLOW: return ERR_BRM_VB_OVERFLOW;
case BRM::ERR_NETWORK: return ERR_BRM_NETWORK;
case BRM::ERR_READONLY: return ERR_BRM_READONLY;
default: return ERR_BRM_BEGIN_COPY;
}
}
VBRange aRange;
uint32_t blocksProcessedThisOid = 0;
uint32_t blocksProcessed = 0;
std::vector<BRM::FileInfo> files;
TableMetaData* aTbaleMetaData = TableMetaData::makeTableMetaData(tableOid);
for (i = 0; i < totalColumn; i++)
{
valArray = NULL;
oldValArray = NULL;
curColStruct = colStructList[i];
curTupleList = colValueList[i]; // same value for all rows
ColumnOp* colOp = m_colOp[op(curColStruct.fCompressionType)];
// convert column data type
if (convertStructFlag)
Convertor::convertColType(&curColStruct);
// set params
colOp->initColumn(curCol);
colOp->setColParam(curCol, 0, curColStruct.colWidth, curColStruct.colDataType, curColStruct.colType,
curColStruct.dataOid, curColStruct.fCompressionType, curColStruct.fColDbRoot,
curColStruct.fColPartition, curColStruct.fColSegment);
colOp->findTypeHandler(curColStruct.colWidth, curColStruct.colDataType);
ColExtsInfo aColExtsInfo = aTbaleMetaData->getColExtsInfo(curColStruct.dataOid);
ColExtsInfo::iterator it = aColExtsInfo.begin();
while (it != aColExtsInfo.end())
{
if ((it->dbRoot == curColStruct.fColDbRoot) && (it->partNum == curColStruct.fColPartition) &&
(it->segNum == curColStruct.fColSegment))
break;
it++;
}
if (it == aColExtsInfo.end()) // add this one to the list
{
ColExtInfo aExt;
aExt.dbRoot = curColStruct.fColDbRoot;
aExt.partNum = curColStruct.fColPartition;
aExt.segNum = curColStruct.fColSegment;
aExt.compType = curColStruct.fCompressionType;
aColExtsInfo.push_back(aExt);
aTbaleMetaData->setColExtsInfo(colStructList[i].dataOid, aColExtsInfo);
}
string segFile;
bool isReadOnly = (m_opType == DELETE && hasAUXCol && (i != colStructList.size() - 1));
rc = colOp->openColumnFile(curCol, segFile, true, IO_BUFF_SIZE, isReadOnly); // @bug 5572 HDFS tmp file
if (rc != NO_ERROR)
break;
if (curColStruct.fCompressionType == 0)
{
BRM::FileInfo aFile;
aFile.oid = curColStruct.dataOid;
aFile.partitionNum = curColStruct.fColPartition;
aFile.dbRoot = curColStruct.fColDbRoot;
aFile.segmentNum = curColStruct.fColSegment;
aFile.compType = curColStruct.fCompressionType;
files.push_back(aFile);
}
// handling versioning
std::vector<VBRange> curFreeList;
uint32_t blockUsed = 0;
if (!idbdatafile::IDBPolicy::useHdfs())
{
if (rangeListTot.size() > 0 &&
(m_opType != DELETE || !hasAUXCol || (i == colStructList.size() - 1)))
{
ColStructList::size_type j = i;
if (m_opType == DELETE && hasAUXCol && (i == colStructList.size() - 1))
j = 0;
if (freeList[0].size >= (blocksProcessed + rangeLists[j].size()))
{
aRange.vbOID = freeList[0].vbOID;
aRange.vbFBO = freeList[0].vbFBO + blocksProcessed;
aRange.size = rangeLists[j].size();
curFreeList.push_back(aRange);
}
else
{
aRange.vbOID = freeList[0].vbOID;
aRange.vbFBO = freeList[0].vbFBO + blocksProcessed;
aRange.size = freeList[0].size - blocksProcessed;
blockUsed = aRange.size;
curFreeList.push_back(aRange);
if (freeList.size() > 1)
{
aRange.vbOID = freeList[1].vbOID;
aRange.vbFBO = freeList[1].vbFBO + blocksProcessedThisOid;
aRange.size = rangeLists[j].size() - blockUsed;
curFreeList.push_back(aRange);
blocksProcessedThisOid += aRange.size;
}
else
{
rc = 1;
break;
}
}
blocksProcessed += rangeLists[j].size();
rc = BRMWrapper::getInstance()->writeVB(curCol.dataFile.pFile, (BRM::VER_t)txnid,
curColStruct.dataOid, fboLists[j], rangeLists[j], colOp,
curFreeList, curColStruct.fColDbRoot, true);
}
}
if (rc != NO_ERROR)
{
if (curColStruct.fCompressionType == 0)
{
curCol.dataFile.pFile->flush();
}
if (rangeListTot.size() > 0)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeListTot);
break;
}
allocateValArray(valArray, 1, curColStruct.colType, curColStruct.colWidth);
ExtCPInfo* cpInfo;
cpInfo = cpInfos ? ((*cpInfos)[i]) : NULL;
if (cpInfo)
{
allocateValArray(oldValArray, totalRow, curColStruct.colType, curColStruct.colWidth);
}
// convert values to valArray
if (m_opType != DELETE)
{
bExcp = false;
ColTuple curTuple;
curTuple = curTupleList[0];
try
{
convertValue(cscColTypeList[i], curColStruct.colType, valArray, curTuple.data);
}
catch (...)
{
bExcp = true;
}
if (bExcp)
{
if (rangeListTot.size() > 0)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeListTot);
return ERR_PARSING;
}
#ifdef PROFILE
timer.start("writeRow ");
#endif
rc = colOp->writeRows(curCol, totalRow, ridList, valArray, oldValArray);
#ifdef PROFILE
timer.stop("writeRow ");
#endif
}
else
{
#ifdef PROFILE
timer.start("writeRows ");
#endif
if (!isReadOnly)
rc = colOp->writeRows(curCol, totalRow, ridList, valArray, oldValArray, true);
else
rc = colOp->writeRowsReadOnly(curCol, totalRow, ridList, oldValArray);
#ifdef PROFILE
timer.stop("writeRows ");
#endif
}
updateMaxMinRange(1, totalRow, cscColTypeList[i], curColStruct.colType,
m_opType == DELETE ? NULL : valArray, oldValArray, cpInfo, false);
colOp->clearColumn(curCol, !isReadOnly);
if (valArray != NULL)
{
free(valArray);
valArray = NULL;
}
if (oldValArray != NULL)
{
free(oldValArray);
oldValArray = NULL;
}
// check error
if (rc != NO_ERROR)
break;
} // end of for (i = 0)
if ((idbdatafile::IDBPolicy::useHdfs()) && (files.size() > 0))
cacheutils::purgePrimProcFdCache(files, Config::getLocalModuleID());
if (rangeListTot.size() > 0)
BRMWrapper::getInstance()->writeVBEnd(txnid, rangeListTot);
// timer.stop("Delete:writecolrec");
#ifdef PROFILE
timer.finish();
#endif
return rc;
}
/*@brief tokenize - return a token for a given signature and size
*/
/***********************************************************
* DESCRIPTION:
* return a token for a given signature and size
* If it is not in the dictionary, the signature
* will be added to the dictionary and the index tree
* If it is already in the dictionary, then
* the token will be returned
* This function does not open and close files.
* users need to use openDctnry and CloseDctnry
* PARAMETERS:
* DctnryTuple& dctnryTuple - holds the sigValue, sigSize and token
* RETURN:
* NO_ERROR if success
* others if something wrong in inserting the value
***********************************************************/
int WriteEngineWrapper::tokenize(const TxnID& txnid, DctnryTuple& dctnryTuple, int ct)
{
int cop = op(ct);
m_dctnry[cop]->setTransId(txnid);
return m_dctnry[cop]->updateDctnry(dctnryTuple.sigValue, dctnryTuple.sigSize, dctnryTuple.token);
}
/*@brief tokenize - return a token for a given signature and size
* accept OIDs as input
*/
/***********************************************************
* DESCRIPTION:
* Token for a given signature and size
* If it is not in the dictionary, the signature
* will be added to the dictionary and the index tree
* If it is already in the dictionary, then
* the token will be returned
* PARAMETERS:
* DctnryTuple& dctnryTuple - holds the sigValue, sigSize and token
* DctnryStruct dctnryStruct- contain the 3 OID for dictionary,
* tree and list.
* RETURN:
* NO_ERROR if success
* others if something wrong in inserting the value
***********************************************************/
int WriteEngineWrapper::tokenize(const TxnID& txnid, DctnryStruct& dctnryStruct, DctnryTuple& dctnryTuple,
bool useTmpSuffix) // @bug 5572 HDFS tmp file
{
// find the corresponding column segment file the token is going to be inserted.
Dctnry* dctnry = m_dctnry[op(dctnryStruct.fCompressionType)];
int rc = dctnry->openDctnry(dctnryStruct.dctnryOid, dctnryStruct.fColDbRoot, dctnryStruct.fColPartition,
dctnryStruct.fColSegment,
useTmpSuffix); // @bug 5572 TBD
if (rc != NO_ERROR)
return rc;
rc = tokenize(txnid, dctnryTuple, dctnryStruct.fCompressionType);
int rc2 = dctnry->closeDctnry(true); // close file, even if tokenize() fails
if ((rc == NO_ERROR) && (rc2 != NO_ERROR))
rc = rc2;
return rc;
}
/***********************************************************
* DESCRIPTION:
* Create column files, including data and bitmap files
* PARAMETERS:
* dataOid - column data file id
* bitmapOid - column bitmap file id
* colWidth - column width
* dbRoot - DBRoot where file is to be located
* partition - Starting partition number for segment file path
* segment - segment number
* compressionType - compression type
* RETURN:
* NO_ERROR if success
* ERR_FILE_EXIST if file exists
* ERR_FILE_CREATE if something wrong in creating the file
***********************************************************/
int WriteEngineWrapper::createDctnry(const TxnID& txnid, const OID& dctnryOid, int colWidth, uint16_t dbRoot,
uint32_t partiotion, uint16_t segment, int compressionType)
{
BRM::LBID_t startLbid;
return m_dctnry[op(compressionType)]->createDctnry(dctnryOid, colWidth, dbRoot, partiotion, segment,
startLbid);
}
int WriteEngineWrapper::convertRidToColumn(RID& rid, uint16_t& dbRoot, uint32_t& partition, uint16_t& segment,
RID filesPerColumnPartition, RID extentsPerSegmentFile,
RID extentRows, uint16_t startDBRoot, unsigned dbrootCnt)
{
int rc = 0;
partition = rid / (filesPerColumnPartition * extentsPerSegmentFile * extentRows);
segment = (((rid % (filesPerColumnPartition * extentsPerSegmentFile * extentRows)) / extentRows)) %
filesPerColumnPartition;
dbRoot = ((startDBRoot - 1 + segment) % dbrootCnt) + 1;
// Calculate the relative rid for this segment file
RID relRidInPartition =
rid - ((RID)partition * (RID)filesPerColumnPartition * (RID)extentsPerSegmentFile * (RID)extentRows);
assert(relRidInPartition <= (RID)filesPerColumnPartition * (RID)extentsPerSegmentFile * (RID)extentRows);
uint32_t numExtentsInThisPart = relRidInPartition / extentRows;
unsigned numExtentsInThisSegPart = numExtentsInThisPart / filesPerColumnPartition;
RID relRidInThisExtent = relRidInPartition - numExtentsInThisPart * extentRows;
rid = relRidInThisExtent + numExtentsInThisSegPart * extentRows;
return rc;
}
/***********************************************************
* DESCRIPTION:
* Clears table lock for the specified table lock ID.
* PARAMETERS:
* lockID - table lock to be released
* errMsg - if error occurs, this is the return error message
* RETURN:
* NO_ERROR if operation is successful
***********************************************************/
int WriteEngineWrapper::clearTableLockOnly(uint64_t lockID, std::string& errMsg)
{
bool bReleased;
int rc = BRMWrapper::getInstance()->releaseTableLock(lockID, bReleased, errMsg);
return rc;
}
/***********************************************************
* DESCRIPTION:
* Rolls back the state of the extentmap and database files for the
* specified table OID, using the metadata previously saved to disk.
* Also clears the table lock for the specified table OID.
* PARAMETERS:
* tableOid - table OID to be rolled back
* lockID - table lock corresponding to tableOid
* tableName - table name associated with tableOid
* applName - application that is driving this bulk rollback
* debugConsole - enable debug logging to the console
* errorMsg - error message explaining any rollback failure
* RETURN:
* NO_ERROR if rollback completed succesfully
***********************************************************/
int WriteEngineWrapper::bulkRollback(OID tableOid, uint64_t lockID, const std::string& tableName,
const std::string& applName, bool debugConsole, string& errorMsg)
{
errorMsg.clear();
BulkRollbackMgr rollbackMgr(tableOid, lockID, tableName, applName);
if (debugConsole)
rollbackMgr.setDebugConsole(true);
// We used to pass "false" to not keep (delete) the metafiles at the end of
// the rollback. But after the transition to sharedNothing, we pass "true"
// to initially keep these files. The metafiles are deleted later, only
// after all the distributed bulk rollbacks are successfully completed.
int rc = rollbackMgr.rollback(true);
if (rc != NO_ERROR)
errorMsg = rollbackMgr.getErrorMsg();
// Ignore the return code for now; more important to base rc on the
// success or failure of the previous work
BRMWrapper::getInstance()->takeSnapshot();
return rc;
}
int WriteEngineWrapper::rollbackCommon(const TxnID& txnid, int sessionId)
{
// Remove the unwanted tmp files and recover compressed chunks.
string prefix;
// BUG 4312
RemoveTxnFromLBIDMap(txnid);
RemoveTxnFromDictMap(txnid);
config::Config* config = config::Config::makeConfig();
prefix = config->getConfig("SystemConfig", "DBRMRoot");
if (prefix.length() == 0)
{
cerr << "Need a valid DBRMRoot entry in Calpont configuation file";
return ERR_INVALID_PARAM;
}
uint64_t pos = prefix.find_last_of("/");
std::string aDMLLogFileName;
if (pos != string::npos)
{
aDMLLogFileName = prefix.substr(0, pos + 1); // Get the file path
}
else
{
logging::Message::Args args;
args.add("RollbackTran cannot find the dbrm directory for the DML log file");
SimpleSysLog::instance()->logMsg(args, logging::LOG_TYPE_CRITICAL, logging::M0007);
return ERR_OPEN_DML_LOG;
}
std::ostringstream oss;
oss << txnid << "_" << Config::getLocalModuleID();
aDMLLogFileName += "DMLLog_" + oss.str();
if (IDBPolicy::exists(aDMLLogFileName.c_str()))
{
// TODO-for now the DML log file will always be in a local
// filesystem since IDBDataFile doesn't have any support for
// a cpp iostream interface. need to decide if this is ok.
boost::scoped_ptr<IDBDataFile> aDMLLogFile(IDBDataFile::open(
IDBPolicy::getType(aDMLLogFileName.c_str(), IDBPolicy::WRITEENG), aDMLLogFileName.c_str(), "r", 0));
if (aDMLLogFile) // need recover
{
ssize_t fileSize = aDMLLogFile->size();
boost::scoped_array<char> buf(new char[fileSize]);
if (aDMLLogFile->read(buf.get(), fileSize) != fileSize)
return ERR_FILE_READ;
std::istringstream strstream(string(buf.get(), fileSize));
std::string backUpFileType;
std::string filename;
int64_t size;
int64_t offset;
while (strstream >> backUpFileType >> filename >> size >> offset)
{
// cout << "Found: " << backUpFileType << " name " << filename << "size: " << size << " offset: " <<
// offset << endl;
std::ostringstream oss;
oss << "RollbackTran found " << backUpFileType << " name " << filename << " size: " << size
<< " offset: " << offset;
logging::Message::Args args;
args.add(oss.str());
SimpleSysLog::instance()->logMsg(args, logging::LOG_TYPE_INFO, logging::M0007);
if (backUpFileType.compare("rlc") == 0)
{
// remove the rlc file
filename += ".rlc";
// cout << " File removed: " << filename << endl;
IDBPolicy::remove(filename.c_str());
logging::Message::Args args1;
args1.add(filename);
args1.add(" is removed.");
SimpleSysLog::instance()->logMsg(args1, logging::LOG_TYPE_INFO, logging::M0007);
}
else if (backUpFileType.compare("tmp") == 0)
{
int rc = NO_ERROR;
string orig(filename + ".orig");
// restore the orig file
if (IDBPolicy::exists(orig.c_str()))
{
// not likely both cdf and tmp exist
if (IDBPolicy::exists(filename.c_str()) && IDBPolicy::remove(filename.c_str()) != 0)
rc = ERR_COMP_REMOVE_FILE;
if (rc == NO_ERROR && IDBPolicy::rename(orig.c_str(), filename.c_str()) != 0)
rc = ERR_COMP_RENAME_FILE;
}
// remove the tmp file
string tmp(filename + ".tmp");
if (rc == NO_ERROR && IDBPolicy::exists(tmp.c_str()) && IDBPolicy::remove(tmp.c_str()) != 0)
rc = ERR_COMP_REMOVE_FILE;
// remove the chunk shifting helper
string rlc(filename + ".rlc");
if (rc == NO_ERROR && IDBPolicy::exists(rlc.c_str()) && IDBPolicy::remove(rlc.c_str()) != 0)
rc = ERR_COMP_REMOVE_FILE;
logging::Message::Args args1;
args1.add(filename);
if (rc == NO_ERROR)
{
args1.add(" is restored.");
SimpleSysLog::instance()->logMsg(args1, logging::LOG_TYPE_INFO, logging::M0007);
}
else
{
args1.add(" may not restored: ");
args1.add(rc);
SimpleSysLog::instance()->logMsg(args1, logging::LOG_TYPE_CRITICAL, logging::M0007);
return rc;
}
}
else
{
// copy back to the data file
std::string backFileName(filename);
if (backUpFileType.compare("chk") == 0)
backFileName += ".chk";
else
backFileName += ".hdr";
// cout << "Rollback found file " << backFileName << endl;
IDBDataFile* sourceFile = IDBDataFile::open(
IDBPolicy::getType(backFileName.c_str(), IDBPolicy::WRITEENG), backFileName.c_str(), "r", 0);
IDBDataFile* targetFile = IDBDataFile::open(
IDBPolicy::getType(filename.c_str(), IDBPolicy::WRITEENG), filename.c_str(), "r+", 0);
size_t byteRead;
unsigned char* readBuf = new unsigned char[size];
boost::scoped_array<unsigned char> readBufPtr(readBuf);
if (sourceFile != NULL)
{
int rc = sourceFile->seek(0, 0);
if (rc)
return ERR_FILE_SEEK;
byteRead = sourceFile->read(readBuf, size);
if ((int)byteRead != size)
{
logging::Message::Args args6;
args6.add("Rollback cannot read backup file ");
args6.add(backFileName);
SimpleSysLog::instance()->logMsg(args6, logging::LOG_TYPE_ERROR, logging::M0007);
return ERR_FILE_READ;
}
}
else
{
logging::Message::Args args5;
args5.add("Rollback cannot open backup file ");
args5.add(backFileName);
SimpleSysLog::instance()->logMsg(args5, logging::LOG_TYPE_ERROR, logging::M0007);
return ERR_FILE_NULL;
}
size_t byteWrite;
if (targetFile != NULL)
{
int rc = targetFile->seek(offset, 0);
if (rc)
return ERR_FILE_SEEK;
byteWrite = targetFile->write(readBuf, size);
if ((int)byteWrite != size)
{
logging::Message::Args args3;
args3.add("Rollback cannot copy to file ");
args3.add(filename);
args3.add("from file ");
args3.add(backFileName);
SimpleSysLog::instance()->logMsg(args3, logging::LOG_TYPE_ERROR, logging::M0007);
return ERR_FILE_WRITE;
}
}
else
{
logging::Message::Args args4;
args4.add("Rollback cannot open target file ");
args4.add(filename);
SimpleSysLog::instance()->logMsg(args4, logging::LOG_TYPE_ERROR, logging::M0007);
return ERR_FILE_NULL;
}
// cout << "Rollback copied to file " << filename << " from file " << backFileName << endl;
delete targetFile;
delete sourceFile;
IDBPolicy::remove(backFileName.c_str());
logging::Message::Args arg1;
arg1.add("Rollback copied to file ");
arg1.add(filename);
arg1.add("from file ");
arg1.add(backFileName);
SimpleSysLog::instance()->logMsg(arg1, logging::LOG_TYPE_INFO, logging::M0007);
}
}
}
IDBPolicy::remove(aDMLLogFileName.c_str());
}
return 0;
}
int WriteEngineWrapper::rollbackTran(const TxnID& txnid, int sessionId)
{
if (rollbackCommon(txnid, sessionId) != 0)
return -1;
return BRMWrapper::getInstance()->rollBack(txnid, sessionId);
}
int WriteEngineWrapper::rollbackBlocks(const TxnID& txnid, int sessionId)
{
if (rollbackCommon(txnid, sessionId) != 0)
return -1;
return BRMWrapper::getInstance()->rollBackBlocks(txnid, sessionId);
}
int WriteEngineWrapper::rollbackVersion(const TxnID& txnid, int sessionId)
{
// BUG 4312
RemoveTxnFromLBIDMap(txnid);
RemoveTxnFromDictMap(txnid);
return BRMWrapper::getInstance()->rollBackVersion(txnid, sessionId);
}
int WriteEngineWrapper::updateNextValue(const TxnID txnId, const OID& columnoid, const uint64_t nextVal,
const uint32_t sessionID, const uint16_t dbRoot)
{
int rc = NO_ERROR;
boost::shared_ptr<CalpontSystemCatalog> systemCatalogPtr;
RIDList ridList;
ColValueList colValueList;
WriteEngine::ColTupleList colTuples;
ColStructList colStructList;
WriteEngine::CSCTypesList cscColTypeList;
WriteEngine::ColStruct colStruct;
CalpontSystemCatalog::ColType colType;
colType.columnOID = colStruct.dataOid = OID_SYSCOLUMN_NEXTVALUE;
colType.colWidth = colStruct.colWidth = 8;
colStruct.tokenFlag = false;
colType.colDataType = colStruct.colDataType = CalpontSystemCatalog::UBIGINT;
colStruct.fColDbRoot = dbRoot;
m_opType = UPDATE;
if (idbdatafile::IDBPolicy::useHdfs())
colStruct.fCompressionType = 2;
colStructList.push_back(colStruct);
cscColTypeList.push_back(colType);
ColTuple colTuple;
systemCatalogPtr = CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
systemCatalogPtr->identity(CalpontSystemCatalog::EC);
CalpontSystemCatalog::ROPair ropair;
try
{
ropair = systemCatalogPtr->nextAutoIncrRid(columnoid);
}
catch (...)
{
rc = ERR_AUTOINC_RID;
}
if (rc != NO_ERROR)
return rc;
ridList.push_back(ropair.rid);
colTuple.data = nextVal;
colTuples.push_back(colTuple);
colValueList.push_back(colTuples);
rc = writeColumnRecords(txnId, cscColTypeList, colStructList, colValueList, ridList, SYSCOLUMN_BASE, false);
if (rc != NO_ERROR)
return rc;
// flush PrimProc cache
vector<LBID_t> blockList;
BRM::LBIDRange_v lbidRanges;
rc = BRMWrapper::getInstance()->lookupLbidRanges(OID_SYSCOLUMN_NEXTVALUE, lbidRanges);
if (rc != NO_ERROR)
return rc;
LBIDRange_v::iterator it;
for (it = lbidRanges.begin(); it != lbidRanges.end(); it++)
{
for (LBID_t lbid = it->start; lbid < (it->start + it->size); lbid++)
{
blockList.push_back(lbid);
}
}
// Bug 5459 Flush FD cache
std::vector<BRM::FileInfo> files;
BRM::FileInfo aFile;
aFile.oid = colStruct.dataOid;
aFile.partitionNum = colStruct.fColPartition;
aFile.dbRoot = colStruct.fColDbRoot;
;
aFile.segmentNum = colStruct.fColSegment;
aFile.compType = colStruct.fCompressionType;
files.push_back(aFile);
if (idbdatafile::IDBPolicy::useHdfs())
cacheutils::purgePrimProcFdCache(files, Config::getLocalModuleID());
rc = cacheutils::flushPrimProcAllverBlocks(blockList);
if (rc != 0)
rc = ERR_BLKCACHE_FLUSH_LIST; // translate to WE error
return rc;
}
/***********************************************************
* DESCRIPTION:
* Flush compressed files in chunk manager
* PARAMETERS:
* none
* RETURN:
* none
***********************************************************/
int WriteEngineWrapper::flushDataFiles(int rc, const TxnID txnId, std::map<FID, FID>& columnOids)
{
RemoveTxnFromLBIDMap(txnId);
RemoveTxnFromDictMap(txnId);
for (int i = 0; i < TOTAL_COMPRESS_OP; i++)
{
int rc1 = m_colOp[i]->flushFile(rc, columnOids);
int rc2 = m_dctnry[i]->flushFile(rc, columnOids);
if (rc == NO_ERROR)
{
rc = (rc1 != NO_ERROR) ? rc1 : rc2;
}
}
return rc;
}
void WriteEngineWrapper::AddDictToList(const TxnID txnid, std::vector<BRM::LBID_t>& lbids)
{
std::tr1::unordered_map<TxnID, dictLBIDRec_t>::iterator mapIter;
mapIter = m_dictLBIDMap.find(txnid);
if (mapIter == m_dictLBIDMap.end())
{
dictLBIDRec_t tempRecord;
tempRecord.insert(lbids.begin(), lbids.end());
m_dictLBIDMap[txnid] = tempRecord;
return;
}
else
{
dictLBIDRec_t& txnRecord = mapIter->second;
txnRecord.insert(lbids.begin(), lbids.end());
}
}
// Get CPInfo for given starting LBID and column description structure.
int WriteEngineWrapper::GetLBIDRange(const BRM::LBID_t startingLBID, const ColStruct& colStruct,
ExtCPInfo& cpInfo)
{
int rtn;
BRM::CPMaxMin maxMin;
rtn = BRMWrapper::getInstance()->getExtentCPMaxMin(startingLBID, maxMin);
bool isBinary = cpInfo.isBinaryColumn();
maxMin.isBinaryColumn = isBinary;
cpInfo.fCPInfo.firstLbid = startingLBID;
if (rtn)
{
cpInfo.toInvalid();
return rtn;
}
// if we are provided with CPInfo pointer to update, we record current range there.
// please note that we may fail here for unknown extents - e.g., newly allocated ones.
// for these we mark CPInfo as invalid (above) and proceed as usual.
// XXX With this logic we may end with invalid ranges for extents that were
// allocated and recorded, yet not in our copy of extentmap.
// As we update only part of that extent, the recorded range will be for
// that part of the extent.
// It should be investigated whether such situation is possible.
// XXX Please note that most if not all calls to AddLBIDToList are enclosed into
// RETURN_ON_ERROR() macro.
// If we have failed to obtain information above we will abort execution of the function
// that called us.
// This is potential source of bugs.
cpInfo.fCPInfo.bigMax = maxMin.bigMax;
cpInfo.fCPInfo.bigMin = maxMin.bigMin;
cpInfo.fCPInfo.max = maxMin.max;
cpInfo.fCPInfo.min = maxMin.min;
cpInfo.fCPInfo.seqNum = maxMin.seqNum;
cpInfo.fCPInfo.isBinaryColumn = maxMin.isBinaryColumn;
return rtn;
}
/***********************************************************
* DESCRIPTION:
* Add an lbid to a list of lbids for sending to markExtentsInvalid.
* However, rather than storing each lbid, store only unique first
* lbids. This is an optimization to prevent invalidating the same
* extents over and over.
* PARAMETERS:
* txnid - the lbid list is per txn. We use this to keep transactions
* separated.
* These next are needed for dbrm to get the lbid:
* colStruct - reference to ColStruct structure of column we record (provides segment and type).
* fbo - file block offset
* cpInfo - a CPInfo to collect if we need that.
* RETURN: 0 => OK. -1 => error
***********************************************************/
int WriteEngineWrapper::AddLBIDtoList(const TxnID txnid, const ColStruct& colStruct, const int fbo,
ExtCPInfo* cpInfo)
{
int rtn = 0;
BRM::LBID_t startingLBID;
SP_TxnLBIDRec_t spTxnLBIDRec;
std::tr1::unordered_map<TxnID, SP_TxnLBIDRec_t>::iterator mapIter;
// Find the set of extent starting LBIDs for this transaction. If not found, then create it.
mapIter = m_txnLBIDMap.find(txnid);
if (mapIter == m_txnLBIDMap.end())
{
// This is a new transaction.
SP_TxnLBIDRec_t sptemp(new TxnLBIDRec);
spTxnLBIDRec = sptemp;
m_txnLBIDMap[txnid] = spTxnLBIDRec;
// cout << "New transaction entry " << txnid << " transaction count " << m_txnLBIDMap.size() <<
// endl;
}
else
{
spTxnLBIDRec = (*mapIter).second;
}
// Get the extent starting lbid given all these values (startingLBID is an out parameter).
rtn = BRMWrapper::getInstance()->getStartLbid(colStruct.dataOid, colStruct.fColPartition,
colStruct.fColSegment, fbo, startingLBID);
if (rtn != 0)
return -1;
// if we are given the cpInfo (column's ranges can be traced), we should update it.
if (cpInfo)
{
rtn = GetLBIDRange(startingLBID, colStruct, *cpInfo);
}
else
{
spTxnLBIDRec->AddLBID(startingLBID, colStruct.colDataType);
}
return rtn;
}
void WriteEngineWrapper::RemoveTxnFromDictMap(const TxnID txnid)
{
std::tr1::unordered_map<TxnID, dictLBIDRec_t>::iterator mapIter;
mapIter = m_dictLBIDMap.find(txnid);
if (mapIter != m_dictLBIDMap.end())
{
m_dictLBIDMap.erase(txnid);
}
}
int WriteEngineWrapper::markTxnExtentsAsInvalid(const TxnID txnid, bool erase)
{
int rc = 0;
std::tr1::unordered_map<TxnID, SP_TxnLBIDRec_t>::iterator mapIter = m_txnLBIDMap.find(txnid);
if (mapIter != m_txnLBIDMap.end())
{
SP_TxnLBIDRec_t spTxnLBIDRec = (*mapIter).second;
if (!spTxnLBIDRec->m_LBIDs.empty())
{
rc = BRMWrapper::getInstance()->markExtentsInvalid(spTxnLBIDRec->m_LBIDs, spTxnLBIDRec->m_ColDataTypes);
}
if (erase)
{
m_txnLBIDMap.erase(txnid); // spTxnLBIDRec is auto-destroyed
}
}
return rc;
}
/***********************************************************
* DESCRIPTION:
* Remove a transaction LBID list from the LBID map
* Called when a transaction ends, either commit or rollback
* PARAMETERS:
* txnid - the transaction to remove.
* RETURN:
* error code
***********************************************************/
int WriteEngineWrapper::RemoveTxnFromLBIDMap(const TxnID txnid)
{
return markTxnExtentsAsInvalid(txnid, true);
}
} // end of namespace
View Git Blame Copy Permalink