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

MCOL-4876 This patch enables continues buffer to be used by ColumnCommand and aligns BPP::blockData

Browse Source 
that in most cases was unaligned
This commit is contained in:
Roman Nozdrin
2021-09-28 08:50:12 +00:00
parent c376472209
commit 3de038c1da
10 changed files with 297 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
primitives/linux-port/column.cpp
View File

@ -926,240 +926,32 @@ inline bool nextColValue(
/// WRITE COLUMN VALUES
///
// Append value to the output buffer with debug-time check for buffer overflow
template<typename T>
inline void checkedWriteValue(
void* out,
unsigned outSize,
unsigned* outPos,
const T* src,
int errSubtype)
{
#ifdef PRIM_DEBUG
if (sizeof(T) > outSize - *outPos)
{
logIt(35, errSubtype);
throw logic_error("PrimitiveProcessor::checkedWriteValue(): output buffer is too small");
}
#endif
uint8_t* out8 = reinterpret_cast<uint8_t*>(out);
memcpy(out8 + *outPos, src, sizeof(T));
*outPos += sizeof(T);
}
// Write the value index in srcArray and/or the value itself, depending on bits in OutputType,
// into the output buffer and update the output pointer.
// TODO Introduce another dispatching layer based on OutputType.
template<typename T>
inline void writeColValue(
uint8_t OutputType,
NewColResultHeader* out,
unsigned outSize,
unsigned* written,
ColResultHeader* out,
uint16_t rid,
const T* srcArray)
{
uint8_t* outPtr = reinterpret_cast<uint8_t*>(&out[1]);
auto idx = out->NVALS++;
if (OutputType & OT_RID)
{
checkedWriteValue(out, outSize, written, &rid, 1);
auto* outPos = getRIDArrayPosition(outPtr, idx);
*outPos = rid;
out->RidFlags |= (1 << (rid >> 9)); // set the (row/512)'th bit
}
if (OutputType & (OT_TOKEN | OT_DATAVALUE))
{
checkedWriteValue(out, outSize, written, &srcArray[rid], 2);
}
out->NVALS++; //TODO: Can be computed at the end from *written value
}
/* WIP
template <bool WRITE_RID, bool WRITE_DATA, bool IS_NULL_VALUE_MATCHES, typename FILTER_ARRAY_T, typename RID_T, typename T>
void writeArray(
size_t dataSize,
const T* dataArray,
const RID_T* dataRid,
const FILTER_ARRAY_T *filterArray,
uint8_t* outbuf,
unsigned* written,
uint16_t* NVALS,
uint8_t* RidFlagsPtr,
T NULL_VALUE)
{
uint8_t* out = outbuf;
uint8_t RidFlags = *RidFlagsPtr;
for (size_t i = 0; i < dataSize; ++i)
{
//TODO: optimize handling of NULL values and flags by avoiding non-predictable jumps
if (dataArray[i]==NULL_VALUE? IS_NULL_VALUE_MATCHES : filterArray[i])
{
if (WRITE_RID)
{
copyValue(out, &dataRid[i], sizeof(RID_T));
out += sizeof(RID_T);
RidFlags |= (1 << (dataRid[i] >> 10)); // set the (row/1024)'th bit
}
if (WRITE_DATA)
{
copyValue(out, &dataArray[i], sizeof(T));
out += sizeof(T);
}
}
}
// Update number of written values, number of written bytes and out->RidFlags
int size1 = (WRITE_RID? sizeof(RID_T) : 0) + (WRITE_DATA? sizeof(T) : 0);
*NVALS += (out - outbuf) / size1;
*written += out - outbuf;
*RidFlagsPtr = RidFlags;
}
*/
/*****************************************************************************
*** RUN DATA THROUGH A COLUMN FILTER ****************************************
*****************************************************************************/
/* "Vertical" processing of the column filter:
1. load all data into temporary vector
2. process one filter element over entire vector before going to a next one
3. write records, that succesfully passed through the filter, to outbuf
*/
/*
template<typename T, ENUM_KIND KIND, typename VALTYPE>
void processArray(
// Source data
const T* srcArray,
size_t srcSize,
uint16_t* ridArray,
size_t ridSize, // Number of values in ridArray
// Filter description
int BOP,
prestored_set_t* filterSet, // Set of values for simple filters (any of values / none of them)
uint32_t filterCount, // Number of filter elements, each described by one entry in the following arrays:
uint8_t* filterCOPs, // comparison operation
int64_t* filterValues, // value to compare to
// Output buffer/stats
uint8_t* outbuf, // Pointer to the place for output data
unsigned* written, // Number of written bytes, that we need to update
uint16_t* NVALS, // Number of written values, that we need to update
uint8_t* RidFlagsPtr, // Pointer to out->RidFlags
// Processing parameters
bool WRITE_RID,
bool WRITE_DATA,
bool SKIP_EMPTY_VALUES,
T EMPTY_VALUE,
bool IS_NULL_VALUE_MATCHES,
T NULL_VALUE,
// Min/Max search
bool ValidMinMax,
VALTYPE* MinPtr,
VALTYPE* MaxPtr)
{
// Alloc temporary arrays
size_t inputSize = (ridArray? ridSize : srcSize);
// Temporary array with data to filter
std::vector<T> dataVec(inputSize);
auto dataArray = dataVec.data();
// Temporary array with RIDs of corresponding dataArray elements
std::vector<RID_T> dataRidVec(WRITE_RID? inputSize : 0);
auto dataRid = dataRidVec.data();
// Copy input data into temporary array, opt. storing RIDs, opt. skipping EMPTYs
size_t dataSize; // number of values copied into dataArray
if (ridArray != NULL)
{
SKIP_EMPTY_VALUES = true; // let findMinMaxArray() know that empty values will be skipped
dataSize = WRITE_RID? readArray<true, true,true>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE)
: readArray<false,true,true>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE);
}
else if (SKIP_EMPTY_VALUES)
{
dataSize = WRITE_RID? readArray<true, false,true>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE)
: readArray<false,false,true>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE);
}
else
{
dataSize = WRITE_RID? readArray<true, false,false>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE)
: readArray<false,false,false>(srcArray, srcSize, dataArray, dataRid, ridArray, ridSize, EMPTY_VALUE);
}
// If required, find Min/Max values of the data
if (ValidMinMax)
{
SKIP_EMPTY_VALUES? findMinMaxArray<true> (dataSize, dataArray, MinPtr, MaxPtr, EMPTY_VALUE, NULL_VALUE)
: findMinMaxArray<false>(dataSize, dataArray, MinPtr, MaxPtr, EMPTY_VALUE, NULL_VALUE);
}
// Choose initial filterArray[i] value depending on the operation
bool initValue = false;
if (filterCount == 0) {initValue = true;}
else if (BOP_NONE == BOP) {initValue = false; BOP = BOP_OR;}
else if (BOP_OR == BOP) {initValue = false;}
else if (BOP_XOR == BOP) {initValue = false;}
else if (BOP_AND == BOP) {initValue = true;}
// Temporary array accumulating results of filtering for each record
std::vector<uint8_t> filterVec(dataSize, initValue);
auto filterArray = filterVec.data();
// Real type of column data, may be floating-point (used only for comparisons in the filtering)
using FLOAT_T = typename std::conditional<sizeof(T) == 8, double, float>::type;
using DATA_T = typename std::conditional<KIND_FLOAT == KIND, FLOAT_T, T>::type;
auto realDataArray = reinterpret_cast<DATA_T*>(dataArray);
// Evaluate column filter on elements of dataArray and store results into filterArray
if (filterSet != NULL && BOP == BOP_OR)
{
applySetFilter<BOP_OR>(dataSize, dataArray, filterSet, filterArray);
}
else if (filterSet != NULL && BOP == BOP_AND)
{
applySetFilter<BOP_AND>(dataSize, dataArray, filterSet, filterArray);
}
else
for (int i = 0; i < filterCount; ++i)
{
DATA_T cmp_value; // value for comparison, may be floating-point
copyValue(&cmp_value, &filterValues[i], sizeof(cmp_value));
switch(BOP)
{
case BOP_AND: applyFilterElement<BOP_AND>(filterCOPs[i], dataSize, realDataArray, cmp_value, filterArray); break;
case BOP_OR: applyFilterElement<BOP_OR> (filterCOPs[i], dataSize, realDataArray, cmp_value, filterArray); break;
case BOP_XOR: applyFilterElement<BOP_XOR>(filterCOPs[i], dataSize, realDataArray, cmp_value, filterArray); break;
default: idbassert(0);
}
}
}
// Copy filtered data and/or their RIDs into output buffer
if (WRITE_RID && WRITE_DATA)
{
IS_NULL_VALUE_MATCHES? writeArray<true,true,true> (dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE)
: writeArray<true,true,false>(dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE);
}
else if (WRITE_RID)
{
IS_NULL_VALUE_MATCHES? writeArray<true,false,true> (dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE)
: writeArray<true,false,false>(dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE);
}
else
{
IS_NULL_VALUE_MATCHES? writeArray<false,true,true> (dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE)
: writeArray<false,true,false>(dataSize, dataArray, dataRid, filterArray, outbuf, written, NVALS, RidFlagsPtr, NULL_VALUE);
T* outPos = getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(out), idx);
// TODO check bytecode for the 16 byte type
*outPos = srcArray[rid];
}
}
*/
// These two are templates update min/max values in the loop iterating the values in filterColumnData.
template<ENUM_KIND KIND, typename T,
@ -1194,9 +986,7 @@ inline void updateMinMax(T& Min, T& Max, T& curValue, NewColRequestHeader* in)
template<typename T, ENUM_KIND KIND>
void filterColumnData(
NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written,
ColResultHeader* out,
uint16_t* ridArray,
const uint16_t ridSize, // Number of values in ridArray
int* srcArray16,
@ -1284,7 +1074,7 @@ void filterColumnData(
{
// If NULL values match the filter, write curValue to the output buffer
if (isNullValueMatches)
writeColValue<T>(outputType, out, outSize, written, rid, srcArray);
writeColValue<T>(outputType, out, rid, srcArray);
}
else
{
@ -1292,7 +1082,7 @@ void filterColumnData(
if (matchingColValue<KIND, COL_WIDTH, false>(curValue, columnFilterMode, filterSet, filterCount,
filterCOPs, filterValues, filterRFs, in->colType, NULL_VALUE))
{
writeColValue<T>(outputType, out, outSize, written, rid, srcArray);
writeColValue<T>(outputType, out, rid, srcArray);
}
// Update Min and Max if necessary. EMPTY/NULL values are processed in other branches.
@ -1347,9 +1137,7 @@ template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int32_t), T>::type* = nullptr>
#endif
void PrimitiveProcessor::scanAndFilterTypeDispatcher(NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written)
ColResultHeader* out)
{
constexpr int W = sizeof(T);
auto dataType = (execplan::CalpontSystemCatalog::ColDataType) in->colType.DataType;
@ -1360,10 +1148,10 @@ void PrimitiveProcessor::scanAndFilterTypeDispatcher(NewColRequestHeader* in,
uint16_t* ridArray = in->getRIDArrayPtr(W);
const uint32_t itemsPerBlock = logicalBlockMode ? BLOCK_SIZE
: BLOCK_SIZE / W;
filterColumnData<T, KIND_FLOAT>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<T, KIND_FLOAT>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
return;
}
_scanAndFilterTypeDispatcher<T>(in, out, outSize, written);
_scanAndFilterTypeDispatcher<T>(in, out);
}
template<typename T,
@ -1377,9 +1165,7 @@ template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int64_t), T>::type* = nullptr>
#endif
void PrimitiveProcessor::scanAndFilterTypeDispatcher(NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written)
ColResultHeader* out)
{
constexpr int W = sizeof(T);
auto dataType = (execplan::CalpontSystemCatalog::ColDataType) in->colType.DataType;
@ -1389,10 +1175,10 @@ void PrimitiveProcessor::scanAndFilterTypeDispatcher(NewColRequestHeader* in,
uint16_t* ridArray = in->getRIDArrayPtr(W);
const uint32_t itemsPerBlock = logicalBlockMode ? BLOCK_SIZE
: BLOCK_SIZE / W;
filterColumnData<T, KIND_FLOAT>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<T, KIND_FLOAT>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
return;
}
_scanAndFilterTypeDispatcher<T>(in, out, outSize, written);
_scanAndFilterTypeDispatcher<T>(in, out);
}
template<typename T,
@ -1408,11 +1194,9 @@ template<typename T,
sizeof(T) == sizeof(int128_t), T>::type* = nullptr>
#endif
void PrimitiveProcessor::scanAndFilterTypeDispatcher(NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written)
ColResultHeader* out)
{
_scanAndFilterTypeDispatcher<T>(in, out, outSize, written);
_scanAndFilterTypeDispatcher<T>(in, out);
}
template<typename T,
@ -1426,9 +1210,7 @@ template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int128_t), T>::type* = nullptr>
#endif
void PrimitiveProcessor::_scanAndFilterTypeDispatcher(NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written)
ColResultHeader* out)
{
constexpr int W = sizeof(T);
const uint16_t ridSize = in->NVALS;
@ -1436,7 +1218,7 @@ void PrimitiveProcessor::_scanAndFilterTypeDispatcher(NewColRequestHeader* in,
const uint32_t itemsPerBlock = logicalBlockMode ? BLOCK_SIZE
: BLOCK_SIZE / W;
filterColumnData<T, KIND_DEFAULT>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<T, KIND_DEFAULT>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
}
template<typename T,
@ -1450,9 +1232,7 @@ template<typename T,
typename std::enable_if<sizeof(T) <= sizeof(int64_t), T>::type* = nullptr>
#endif
void PrimitiveProcessor::_scanAndFilterTypeDispatcher(NewColRequestHeader* in,
NewColResultHeader* out,
unsigned outSize,
unsigned* written)
ColResultHeader* out)
{
constexpr int W = sizeof(T);
const uint16_t ridSize = in->NVALS;
@ -1466,24 +1246,23 @@ void PrimitiveProcessor::_scanAndFilterTypeDispatcher(NewColRequestHeader* in,
dataType == execplan::CalpontSystemCatalog::TEXT) &&
!isDictTokenScan(in))
{
filterColumnData<T, KIND_TEXT>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<T, KIND_TEXT>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
return;
}
if (datatypes::isUnsigned(dataType))
{
using UT = typename std::conditional<std::is_unsigned<T>::value || datatypes::is_uint128_t<T>::value, T, typename datatypes::make_unsigned<T>::type>::type;
filterColumnData<UT, KIND_UNSIGNED>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<UT, KIND_UNSIGNED>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
return;
}
filterColumnData<T, KIND_DEFAULT>(in, out, outSize, written, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
filterColumnData<T, KIND_DEFAULT>(in, out, ridArray, ridSize, block, itemsPerBlock, parsedColumnFilter);
}
// The entrypoint for block scanning and filtering.
// The block is in in msg, out msg is used to store values|RIDs matched.
template<typename T>
void PrimitiveProcessor::columnScanAndFilter(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written)
void PrimitiveProcessor::columnScanAndFilter(NewColRequestHeader* in, ColResultHeader* out)
{
#ifdef PRIM_DEBUG
auto markEvent = [&] (char eventChar)
@ -1501,7 +1280,6 @@ void PrimitiveProcessor::columnScanAndFilter(NewColRequestHeader* in, NewColResu
out->ism.Command = COL_RESULTS;
out->OutputType = in->OutputType;
out->RidFlags = 0;
*written = sizeof(NewColResultHeader);
//...Initialize I/O counts;
out->CacheIO = 0;
out->PhysicalIO = 0;
@ -1523,22 +1301,22 @@ void PrimitiveProcessor::columnScanAndFilter(NewColRequestHeader* in, NewColResu
// Sort ridArray (the row index array) if there are RIDs with this in msg
in->sortRIDArrayIfNeeded(W);
scanAndFilterTypeDispatcher<T>(in, out, outSize, written);
scanAndFilterTypeDispatcher<T>(in, out);
#ifdef PRIM_DEBUG
markEvent('C');
#endif
}
template
void primitives::PrimitiveProcessor::columnScanAndFilter<int8_t>(NewColRequestHeader*, NewColResultHeader*, unsigned, unsigned*);
void primitives::PrimitiveProcessor::columnScanAndFilter<int8_t>(NewColRequestHeader*, ColResultHeader*);
template
void primitives::PrimitiveProcessor::columnScanAndFilter<int16_t>(NewColRequestHeader*, NewColResultHeader*, unsigned int, unsigned int*);
void primitives::PrimitiveProcessor::columnScanAndFilter<int16_t>(NewColRequestHeader*, ColResultHeader*);
template
void primitives::PrimitiveProcessor::columnScanAndFilter<int32_t>(NewColRequestHeader*, NewColResultHeader*, unsigned int, unsigned int*);
void primitives::PrimitiveProcessor::columnScanAndFilter<int32_t>(NewColRequestHeader*, ColResultHeader*);
template
void primitives::PrimitiveProcessor::columnScanAndFilter<int64_t>(NewColRequestHeader*, NewColResultHeader*, unsigned int, unsigned int*);
void primitives::PrimitiveProcessor::columnScanAndFilter<int64_t>(NewColRequestHeader*, ColResultHeader*);
template
void primitives::PrimitiveProcessor::columnScanAndFilter<int128_t>(NewColRequestHeader*, NewColResultHeader*, unsigned int, unsigned int*);
void primitives::PrimitiveProcessor::columnScanAndFilter<int128_t>(NewColRequestHeader*, ColResultHeader*);
} // namespace primitives
// vim:ts=4 sw=4:

114
primitives/linux-port/pp-scan-unittest.cpp
View File

@ -2472,7 +2472,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
uint8_t* results;
int fd;
uint32_t i, written;
@ -2506,7 +2506,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 1;
in->DataType = CalpontSystemCatalog::CHAR;
@ -2518,7 +2518,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = &output[sizeof(NewColResultHeader)];
results = &output[sizeof(ColResultHeader)];
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 8160);
@ -2536,7 +2536,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
uint16_t* results;
uint32_t written, i;
int fd;
@ -2570,7 +2570,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 2;
in->DataType = CalpontSystemCatalog::INT;
@ -2582,7 +2582,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<uint16_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<uint16_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 4096);
@ -2600,7 +2600,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
uint32_t* results;
uint32_t written, i;
int fd;
@ -2634,7 +2634,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 4;
in->DataType = CalpontSystemCatalog::INT;
@ -2646,7 +2646,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<uint32_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<uint32_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 2048);
@ -2664,7 +2664,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
u_int64_t* results;
uint32_t written, i;
int fd;
@ -2698,7 +2698,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 8;
in->DataType = CalpontSystemCatalog::INT;
@ -2710,7 +2710,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 1024);
@ -2728,7 +2728,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
uint8_t* results;
uint16_t* rids;
uint32_t written, i;
@ -2763,7 +2763,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
rids = reinterpret_cast<uint16_t*>(&in[1]);
in->DataSize = 1;
@ -2778,7 +2778,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<uint8_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<uint8_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 2);
@ -2796,7 +2796,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
uint32_t* results;
uint32_t written, i;
@ -2831,7 +2831,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&in[1]);
in->DataSize = 4;
@ -2854,7 +2854,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<uint32_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<uint32_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 9);
@ -2872,7 +2872,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
u_int64_t* results;
uint32_t written, i;
@ -2908,7 +2908,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 8;
@ -2931,7 +2931,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 33);
@ -2949,7 +2949,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
u_int64_t* results;
uint32_t written, i;
@ -2985,7 +2985,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 8;
@ -3008,7 +3008,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 2);
CPPUNIT_ASSERT(results[0] == 10);
@ -3026,7 +3026,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
uint16_t* rids;
u_int64_t* results;
@ -3063,7 +3063,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 8;
@ -3092,7 +3092,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<u_int64_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 1);
CPPUNIT_ASSERT(results[0] == 10);
@ -3109,7 +3109,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
int16_t* results;
uint32_t written, i;
@ -3145,7 +3145,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 8;
@ -3168,7 +3168,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<int16_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<int16_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 33);
@ -3186,7 +3186,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
int16_t* resultRid;
int64_t* resultVal;
@ -3223,7 +3223,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 8;
@ -3252,7 +3252,7 @@ public:
for (i = 0; i < out->NVALS; i++)
{
resultRid = reinterpret_cast<int16_t*>(&output[
sizeof(NewColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
sizeof(ColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
resultVal = reinterpret_cast<int64_t*>(&resultRid[1]);
// cout << i << ": rid:" << (int) *resultRid << " val:" << *resultVal << endl;
@ -3268,7 +3268,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
uint8_t* results;
uint32_t written, i;
@ -3303,7 +3303,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 1;
in->DataType = CalpontSystemCatalog::CHAR;
@ -3326,7 +3326,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = &output[sizeof(NewColResultHeader)];
results = &output[sizeof(ColResultHeader)];
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 32);
@ -3346,7 +3346,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
int16_t* results;
uint32_t written, i;
@ -3383,7 +3383,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = 4;
@ -3420,7 +3420,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<int16_t*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<int16_t*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 2);
CPPUNIT_ASSERT(results[0] == 8);
@ -3438,7 +3438,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
double* results;
uint32_t written, i;
@ -3474,7 +3474,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 8;
in->DataType = CalpontSystemCatalog::DOUBLE;
@ -3496,7 +3496,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<double*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<double*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 8);
@ -3514,7 +3514,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
float* resultVal;
uint32_t written, i;
@ -3551,7 +3551,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 4;
in->DataType = CalpontSystemCatalog::FLOAT;
@ -3579,7 +3579,7 @@ public:
for (i = 0; i < out->NVALS; i++)
{
resultRid = reinterpret_cast<int16_t*>(&output[
sizeof(NewColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
sizeof(ColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
resultVal = reinterpret_cast<float*>(&resultRid[1]);
// cout << i << ": rid:" << (int) *resultRid << " val:" << *resultVal << endl;
@ -3595,7 +3595,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
float* resultVal;
int16_t* resultRid;
@ -3632,7 +3632,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 4;
in->DataType = CalpontSystemCatalog::FLOAT;
@ -3661,7 +3661,7 @@ public:
for (i = 0; i < out->NVALS; i++)
{
resultRid = reinterpret_cast<int16_t*>(&output[
sizeof(NewColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
sizeof(ColResultHeader) + i * (sizeof(Int16) + in->DataSize)]);
resultVal = reinterpret_cast<float*>(&resultRid[1]);
// cout << i << ": rid:" << (int) *resultRid << " val:" << *resultVal << endl;
@ -3677,7 +3677,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
double* results;
uint32_t written, i;
@ -3713,7 +3713,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
in->DataSize = 8;
in->DataType = CalpontSystemCatalog::DOUBLE;
@ -3735,7 +3735,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<double*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<double*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT(out->NVALS == 19);
@ -3764,7 +3764,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[BLOCK_SIZE], output[4 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
binary16* results;
uint32_t written, i;
@ -3797,7 +3797,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = sizeof(binary16);
@ -3825,7 +3825,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<binary16*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<binary16*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT_EQUAL((uint16_t)3, out->NVALS);
CPPUNIT_ASSERT_EQUAL((u_int64_t)10, results[0].uint64(0));
@ -3841,7 +3841,7 @@ public:
PrimitiveProcessor pp;
uint8_t input[2 * BLOCK_SIZE], output[8 * BLOCK_SIZE], block[BLOCK_SIZE];
NewColRequestHeader* in;
NewColResultHeader* out;
ColResultHeader* out;
ColArgs* args;
binary32* results;
uint32_t written, i;
@ -3876,7 +3876,7 @@ public:
memset(output, 0, 4 * BLOCK_SIZE);
in = reinterpret_cast<NewColRequestHeader*>(input);
out = reinterpret_cast<NewColResultHeader*>(output);
out = reinterpret_cast<ColResultHeader*>(output);
args = reinterpret_cast<ColArgs*>(&input[sizeof(NewColRequestHeader)]);
in->DataSize = sizeof(binary32);
@ -3907,7 +3907,7 @@ public:
pp.setBlockPtr((int*) block);
pp.p_Col(in, out, 4 * BLOCK_SIZE, &written);
results = reinterpret_cast<binary32*>(&output[sizeof(NewColResultHeader)]);
results = reinterpret_cast<binary32*>(&output[sizeof(ColResultHeader)]);
// cout << "NVALS = " << out->NVALS << endl;
CPPUNIT_ASSERT_EQUAL((uint16_t)3, out->NVALS);
// CPPUNIT_ASSERT_EQUAL((u_int64_t)10, results[0].uint64(0));

22
primitives/linux-port/primitiveprocessor.h
View File

@ -376,7 +376,7 @@ public:
* an array of 'NOPS' defining the filter to apply (optional),
* followed by an array of RIDs to apply the filter to (optional).
* @param out The buffer that will contain the results. On return, it will start with
* a NewColResultHeader, followed by the output type specified by in->OutputType.
* a ColResultHeader, followed by the output type specified by in->OutputType.
* \li If OT_RID, it will be an array of RIDs
* \li If OT_DATAVALUE, it will be an array of matching data values stored in the column
* \li If OT_BOTH, it will be an array of <RID, DataValue> pairs
@ -385,38 +385,32 @@ public:
* number of bytes written to out.
* @note See PrimitiveMsg.h for the type definitions.
*/
void p_Col(NewColRequestHeader* in, NewColResultHeader* out, unsigned outSize,
void p_Col(NewColRequestHeader* in, ColResultHeader* out, unsigned outSize,
unsigned* written);
template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int8_t) ||
sizeof(T) == sizeof(int16_t) ||
sizeof(T) == sizeof(int128_t), T>::type* = nullptr>
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, ColResultHeader* out);
template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int32_t), T>::type* = nullptr>
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, ColResultHeader* out);
template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int64_t), T>::type* = nullptr>
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void scanAndFilterTypeDispatcher(NewColRequestHeader* in, ColResultHeader* out);
template<typename T,
typename std::enable_if<sizeof(T) <= sizeof(int64_t), T>::type* = nullptr>
void _scanAndFilterTypeDispatcher(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void _scanAndFilterTypeDispatcher(NewColRequestHeader* in, ColResultHeader* out);
template<typename T,
typename std::enable_if<sizeof(T) == sizeof(int128_t), T>::type* = nullptr>
void _scanAndFilterTypeDispatcher(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void _scanAndFilterTypeDispatcher(NewColRequestHeader* in, ColResultHeader* out);
template<typename T>
void columnScanAndFilter(NewColRequestHeader* in, NewColResultHeader* out,
unsigned outSize, unsigned* written);
void columnScanAndFilter(NewColRequestHeader* in, ColResultHeader* out);
boost::shared_ptr<ParsedColumnFilter> parseColumnFilter(const uint8_t* filterString,
uint32_t colWidth,

3
primitives/primproc/batchprimitiveprocessor.cpp
View File

@ -37,6 +37,7 @@
#include <string>
#include <sstream>
#include <set>
#include <stdlib.h>
using namespace std;
#include <boost/thread.hpp>
@ -919,7 +920,7 @@ void BatchPrimitiveProcessor::initProcessor()
strValues.reset(new string[LOGICAL_BLOCK_RIDS]);
outMsgSize = defaultBufferSize;
outputMsg.reset(new uint8_t[outMsgSize]);
outputMsg.reset(reinterpret_cast<uint8_t*>(aligned_alloc(utils::MAXCOLUMNWIDTH, outMsgSize)));
if (ot == ROW_GROUP)
{

2
primitives/primproc/batchprimitiveprocessor.h
View File

@ -225,7 +225,7 @@ private:
uint16_t wideColumnsWidths;
/* Common space for primitive data */
uint8_t blockData[BLOCK_SIZE * utils::MAXCOLUMNWIDTH];
alignas(utils::MAXCOLUMNWIDTH) uint8_t blockData[BLOCK_SIZE * utils::MAXCOLUMNWIDTH];
boost::scoped_array<uint8_t> outputMsg;
uint32_t outMsgSize;

136
primitives/primproc/columncommand.cpp
View File

@ -310,8 +310,9 @@ template<int W>
void ColumnCommand::_issuePrimitive()
{
using IntegralType = typename datatypes::WidthToSIntegralType<W>::type;
uint32_t resultSize;
bpp->getPrimitiveProcessor().columnScanAndFilter<IntegralType>(primMsg, outMsg, bpp->getOutMsgSize(), (unsigned int*)&resultSize);
// Down the call stack the code presumes outMsg buffer has enough space to store
// ColRequestHeader + uint16_t Rids[8192] + IntegralType[8192].
bpp->getPrimitiveProcessor().columnScanAndFilter<IntegralType>(primMsg, outMsg);
} // _issuePrimitive()
void ColumnCommand::updateCPDataNarrow()
@ -360,34 +361,33 @@ void ColumnCommand::_process_OT_BOTH_wAbsRids()
using T = typename datatypes::WidthToSIntegralType<W>::type;
bpp->ridCount = outMsg->NVALS;
bpp->ridMap = outMsg->RidFlags;
size_t pos = sizeof(NewColResultHeader);
uint8_t* outPtr = reinterpret_cast<uint8_t*>(&outMsg[1]);
auto* ridPos = primitives::getRIDArrayPosition(outPtr, 0);
T* valuesPos = primitives::getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(outMsg), 0);
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
bpp->absRids[i] = *((uint16_t*) &bpp->outputMsg[pos]) + bpp->baseRid;
bpp->relRids[i] = *((uint16_t*) &bpp->outputMsg[pos]);
pos += 2;
values[i] = *((T*) &bpp->outputMsg[pos]);
pos += W;
bpp->relRids[i] = ridPos[i];
bpp->absRids[i] = ridPos[i] + bpp->baseRid;
values[i] = valuesPos[i];
}
}
template<>
void ColumnCommand::_process_OT_BOTH_wAbsRids<16>()
{
using T = typename datatypes::WidthToSIntegralType<16>::type;
bpp->ridCount = outMsg->NVALS;
bpp->ridMap = outMsg->RidFlags;
size_t pos = sizeof(NewColResultHeader);
uint8_t* outPtr = reinterpret_cast<uint8_t*>(&outMsg[1]);
auto* ridPos = primitives::getRIDArrayPosition(outPtr, 0);
int128_t* valuesPos = primitives::getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(outMsg), 0);
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
bpp->absRids[i] = *((uint16_t*) &bpp->outputMsg[pos]) + bpp->baseRid;
bpp->relRids[i] = *((uint16_t*) &bpp->outputMsg[pos]);
pos += 2;
datatypes::TSInt128::assignPtrPtr(&wide128Values[i], &bpp->outputMsg[pos]);
pos += 16;
bpp->relRids[i] = ridPos[i];
bpp->absRids[i] = ridPos[i] + bpp->baseRid;
datatypes::TSInt128::assignPtrPtr(&wide128Values[i], &valuesPos[i]);
}
}
@ -399,30 +399,31 @@ void ColumnCommand::_process_OT_BOTH()
bpp->ridCount = outMsg->NVALS;
bpp->ridCount = outMsg->NVALS;
bpp->ridMap = outMsg->RidFlags;
size_t pos = sizeof(NewColResultHeader);
uint8_t* outPtr = reinterpret_cast<uint8_t*>(&outMsg[1]);
auto* ridPos = primitives::getRIDArrayPosition(outPtr, 0);
T* valuesPos = primitives::getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(outMsg), 0);
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
bpp->relRids[i] = *((uint16_t*) &bpp->outputMsg[pos]);
pos += 2;
values[i] = *((T*) &bpp->outputMsg[pos]);
pos += W;
bpp->relRids[i] = ridPos[i];
values[i] = valuesPos[i];
}
}
template<>
void ColumnCommand::_process_OT_BOTH<16>()
{
using T = typename datatypes::WidthToSIntegralType<16>::type;
bpp->ridCount = outMsg->NVALS;
bpp->ridMap = outMsg->RidFlags;
size_t pos = sizeof(NewColResultHeader);
uint8_t* outPtr = reinterpret_cast<uint8_t*>(&outMsg[1]);
auto* ridPos = primitives::getRIDArrayPosition(outPtr, 0);
T* valuesPos = primitives::getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(outMsg), 0);
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
bpp->relRids[i] = *((uint16_t*) &bpp->outputMsg[pos]);
pos += 2;
datatypes::TSInt128::assignPtrPtr(&wide128Values[i], &bpp->outputMsg[pos]);
pos += 16;
bpp->relRids[i] = ridPos[i];
datatypes::TSInt128::assignPtrPtr(&wide128Values[i], &valuesPos[i]);
}
}
@ -487,6 +488,8 @@ void ColumnCommand::process_OT_RID()
bpp->ridMap = outMsg->RidFlags;
}
// TODO This spec makes an impicit type conversion to fit types with sizeof(type) <= 4
// into 8 bytes. Treat values as a buffer and use memcpy to store the values in one go.
template<int W>
void ColumnCommand::_process_OT_DATAVALUE()
{
@ -689,18 +692,17 @@ void ColumnCommand::prep(int8_t outputType, bool absRids)
void ColumnCommand::fillInPrimitiveMessageHeader(const int8_t outputType, const bool absRids)
{
// WIP Align this structure or move input RIDs away.
baseMsgLength = sizeof(NewColRequestHeader) +
(suppressFilter ? 0 : filterString.length());
size_t inputMsgBufSize = baseMsgLength + (LOGICAL_BLOCK_RIDS * sizeof(primitives::RIDType));
if (!inputMsg)
inputMsg.reset(new uint8_t[baseMsgLength + (LOGICAL_BLOCK_RIDS * 2)]);
inputMsg.reset(reinterpret_cast<uint8_t*>(aligned_alloc(utils::MAXCOLUMNWIDTH,
inputMsgBufSize)));
primMsg = (NewColRequestHeader*) inputMsg.get();
outMsg = (NewColResultHeader*) bpp->outputMsg.get();
makeAbsRids = absRids;
primMsg = (NewColRequestHeader*) inputMsg.get();
outMsg = (NewColResultHeader*) bpp->outputMsg.get();
outMsg = (ColResultHeader*) bpp->outputMsg.get();
makeAbsRids = absRids;
primMsg->ism.Interleave = 0;
primMsg->ism.Flags = 0;
@ -721,7 +723,8 @@ void ColumnCommand::fillInPrimitiveMessageHeader(const int8_t outputType, const
/* Assumes OT_DATAVALUE */
void ColumnCommand::projectResult()
{
if (primMsg->NVALS != outMsg->NVALS || outMsg->NVALS != bpp->ridCount )
auto nvals = outMsg->NVALS;
if (primMsg->NVALS != nvals || nvals != bpp->ridCount )
{
ostringstream os;
BRM::DBRM brm;
@ -739,19 +742,20 @@ void ColumnCommand::projectResult()
else
os << ": ridcount " << bpp->ridCount;
os << ", output rids " << outMsg->NVALS << endl;
os << ", output rids " << nvals << endl;
//cout << os.str();
if (bpp->sessionID & 0x80000000)
throw NeedToRestartJob(os.str());
else
throw PrimitiveColumnProjectResultExcept(os.str());
throw PrimitiveColumnProjectResultExcept(os.str());
}
idbassert(primMsg->NVALS == outMsg->NVALS);
idbassert(outMsg->NVALS == bpp->ridCount);
*bpp->serialized << (uint32_t) (outMsg->NVALS * colType.colWidth);
bpp->serialized->append((uint8_t*) (outMsg + 1), outMsg->NVALS * colType.colWidth);
idbassert(primMsg->NVALS == nvals);
idbassert(bpp->ridCount == nvals);
uint32_t valuesByteSize = nvals * colType.colWidth;
*bpp->serialized << valuesByteSize;
bpp->serialized->append(primitives::getFirstValueArrayPosition(outMsg), valuesByteSize);
}
void ColumnCommand::removeRowsFromRowGroup(RowGroup& rg)
@ -796,30 +800,26 @@ void ColumnCommand::removeRowsFromRowGroup(RowGroup& rg)
primMsg->NVALS = outMsg->NVALS;
}
template<int W>
template<typename T>
void ColumnCommand::_projectResultRGLoop(rowgroup::Row& r,
uint8_t* msg8,
const uint32_t gapSize,
const uint32_t offset)
const T* valuesArray,
const uint32_t offset)
{
using T = typename datatypes::WidthToSIntegralType<W>::type;
for (size_t i = 0; i < outMsg->NVALS; ++i, msg8 += gapSize)
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
r.setUintField_offset<W>(*((T*) msg8), offset);
r.setIntField_offset(valuesArray[i], offset);
r.nextRow(rowSize);
}
}
template<>
void ColumnCommand::_projectResultRGLoop<16>(rowgroup::Row& r,
uint8_t* msg8,
const uint32_t gapSize,
const uint32_t offset)
void ColumnCommand::_projectResultRGLoop<int128_t>(rowgroup::Row& r,
const int128_t* valuesArray,
const uint32_t offset)
{
using T = typename datatypes::WidthToSIntegralType<16>::type;
for (size_t i = 0; i < outMsg->NVALS; ++i, msg8 += gapSize)
for (size_t i = 0; i < outMsg->NVALS; ++i)
{
r.setBinaryField_offset((T*)msg8, colType.colWidth, offset);
r.setBinaryField_offset(&valuesArray[i], colType.colWidth, offset);
r.nextRow(rowSize);
}
}
@ -827,24 +827,16 @@ void ColumnCommand::_projectResultRGLoop<16>(rowgroup::Row& r,
template<int W>
void ColumnCommand::_projectResultRG(RowGroup& rg, uint32_t pos)
{
uint32_t offset, gapSize;
uint8_t* msg8 = (uint8_t*) (outMsg + 1);
using T = typename datatypes::WidthToSIntegralType<W>::type;
T* valuesArray = primitives::getValuesArrayPosition<T>(primitives::getFirstValueArrayPosition(outMsg), 0);
uint32_t offset;
auto nvals = outMsg->NVALS;
if (noVB)
{
// outMsg has rids in this case
msg8 += 2;
gapSize = colType.colWidth + 2;
}
else
gapSize = colType.colWidth;
/* TODO: reoptimize these away */
rg.initRow(&r);
offset = r.getOffset(pos);
rowSize = r.getSize();
if ((primMsg->NVALS != outMsg->NVALS || outMsg->NVALS != bpp->ridCount) && (!noVB || bpp->sessionID & 0x80000000))
if ((primMsg->NVALS != nvals || nvals != bpp->ridCount) && (!noVB || bpp->sessionID & 0x80000000))
{
ostringstream os;
BRM::DBRM brm;
@ -857,12 +849,12 @@ void ColumnCommand::_projectResultRG(RowGroup& rg, uint32_t pos)
os << __FILE__ << " error on projectResultRG for oid " << oid << " lbid " << lbid;
if (primMsg->NVALS != outMsg->NVALS )
if (primMsg->NVALS != nvals )
os << ": input rids " << primMsg->NVALS;
else
os << ": ridcount " << bpp->ridCount;
os << ", output rids " << outMsg->NVALS;
os << ", output rids " << nvals;
os << endl;
@ -871,14 +863,14 @@ void ColumnCommand::_projectResultRG(RowGroup& rg, uint32_t pos)
else
throw PrimitiveColumnProjectResultExcept(os.str());
}
else if (primMsg->NVALS != outMsg->NVALS || outMsg->NVALS != bpp->ridCount)
else if (primMsg->NVALS != nvals || nvals != bpp->ridCount)
removeRowsFromRowGroup(rg);
idbassert(primMsg->NVALS == outMsg->NVALS);
idbassert(outMsg->NVALS == bpp->ridCount);
idbassert(primMsg->NVALS == nvals);
idbassert(bpp->ridCount == nvals);
rg.getRow(0, &r);
_projectResultRGLoop<W>(r, msg8, gapSize, offset);
_projectResultRGLoop(r, valuesArray, offset);
}
void ColumnCommand::projectResultRG(RowGroup& rg, uint32_t pos)

7
primitives/primproc/columncommand.h
View File

@ -143,10 +143,9 @@ protected:
void _process_OT_DATAVALUE();
void process_OT_ROWGROUP();
void projectResult();
template<int W>
template<typename T>
void _projectResultRGLoop(rowgroup::Row& r,
uint8_t* msg8,
const uint32_t gapSize,
const T* valuesArray,
const uint32_t offset);
template<int W>
void _projectResultRG(rowgroup::RowGroup& rg, uint32_t pos);
@ -164,7 +163,7 @@ protected:
boost::scoped_array<uint8_t> inputMsg;
NewColRequestHeader* primMsg;
NewColResultHeader* outMsg;
ColResultHeader* outMsg;
// the length of base prim msg, which is everything up to the
// rid array for the pCol message