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

Merge pull request #1842 from denis0x0D/MCOL-987_LZ

Browse Source 
MCOL-987 LZ4 compression support.
This commit is contained in:
Roman Nozdrin
2021-07-07 13:13:18 +03:00
committed by GitHub
parent 7b4f759592 cc1c3629c5
commit 866dc25729
45 changed files with 1311 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
writeengine/bulk/we_bulkload.cpp
View File

@ -337,15 +337,12 @@ int BulkLoad::loadJobInfo(
}
}
// Validate that specified compression type is available
compress::IDBCompressInterface compressor;
for (unsigned kT = 0; kT < curJob.jobTableList.size(); kT++)
{
for (unsigned kC = 0; kC < curJob.jobTableList[kT].colList.size(); kC++)
{
if ( !compressor.isCompressionAvail(
curJob.jobTableList[kT].colList[kC].compressionType) )
if (!compress::CompressInterface::isCompressionAvail(
curJob.jobTableList[kT].colList[kC].compressionType))
{
std::ostringstream oss;
oss << "Specified compression type (" <<

108
writeengine/bulk/we_colbufcompressed.cpp
View File

@ -60,12 +60,11 @@ ColumnBufferCompressed::ColumnBufferCompressed( ColumnInfo* pColInfo,
fToBeCompressedBuffer(0),
fToBeCompressedCapacity(0),
fNumBytes(0),
fCompressor(0),
fPreLoadHWMChunk(true),
fFlushedStartHwmChunk(false)
{
fUserPaddingBytes = Config::getNumCompressedPadBlks() * BYTE_PER_BLOCK;
fCompressor = new compress::IDBCompressInterface( fUserPaddingBytes );
compress::initializeCompressorPool(fCompressorPool, fUserPaddingBytes);
}
//------------------------------------------------------------------------------
@ -79,7 +78,6 @@ ColumnBufferCompressed::~ColumnBufferCompressed()
fToBeCompressedBuffer = 0;
fToBeCompressedCapacity = 0;
fNumBytes = 0;
delete fCompressor;
}
//------------------------------------------------------------------------------
@ -91,9 +89,7 @@ int ColumnBufferCompressed::setDbFile(IDBDataFile* f, HWM startHwm, const char*
fFile = f;
fStartingHwm = startHwm;
IDBCompressInterface compressor;
if (compressor.getPtrList(hdrs, fChunkPtrs) != 0)
if (compress::CompressInterface::getPtrList(hdrs, fChunkPtrs) != 0)
{
return ERR_COMP_PARSE_HDRS;
}
@ -102,7 +98,15 @@ int ColumnBufferCompressed::setDbFile(IDBDataFile* f, HWM startHwm, const char*
// rollback), that fall after the HWM, then drop those trailing ptrs.
unsigned int chunkIndex = 0;
unsigned int blockOffsetWithinChunk = 0;
fCompressor->locateBlock(fStartingHwm, chunkIndex, blockOffsetWithinChunk);
auto compressor = compress::getCompressorByType(
fCompressorPool, fColInfo->column.compressionType);
if (!compressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
compressor->locateBlock(fStartingHwm, chunkIndex, blockOffsetWithinChunk);
if ((chunkIndex + 1) < fChunkPtrs.size())
{
@ -127,11 +131,11 @@ int ColumnBufferCompressed::resetToBeCompressedColBuf(
if (!fToBeCompressedBuffer)
{
fToBeCompressedBuffer =
new unsigned char[IDBCompressInterface::UNCOMPRESSED_INBUF_LEN];
new unsigned char[CompressInterface::UNCOMPRESSED_INBUF_LEN];
}
BlockOp::setEmptyBuf( fToBeCompressedBuffer,
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN,
CompressInterface::UNCOMPRESSED_INBUF_LEN,
fColInfo->column.emptyVal,
fColInfo->column.width );
@ -147,10 +151,10 @@ int ColumnBufferCompressed::resetToBeCompressedColBuf(
fLog->logMsg( oss.str(), MSGLVL_INFO2 );
}
fToBeCompressedCapacity = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
fToBeCompressedCapacity = CompressInterface::UNCOMPRESSED_INBUF_LEN;
// Set file offset past end of last chunk
startFileOffset = IDBCompressInterface::HDR_BUF_LEN * 2;
startFileOffset = CompressInterface::HDR_BUF_LEN * 2;
if (fChunkPtrs.size() > 0)
startFileOffset = fChunkPtrs[ fChunkPtrs.size() - 1 ].first +
@ -223,7 +227,7 @@ int ColumnBufferCompressed::writeToFile(int startOffset, int writeSize,
// Expand the compression buffer size if working with an abbrev extent, and
// the bytes we are about to add will overflow the abbreviated extent.
if ((fToBeCompressedCapacity < IDBCompressInterface::UNCOMPRESSED_INBUF_LEN) &&
if ((fToBeCompressedCapacity < CompressInterface::UNCOMPRESSED_INBUF_LEN) &&
((fNumBytes + writeSize + fillUpWEmptiesWriteSize) > fToBeCompressedCapacity) )
{
std::ostringstream oss;
@ -233,7 +237,7 @@ int ColumnBufferCompressed::writeToFile(int startOffset, int writeSize,
"; part-" << fColInfo->curCol.dataFile.fPartition <<
"; seg-" << fColInfo->curCol.dataFile.fSegment;
fLog->logMsg( oss.str(), MSGLVL_INFO2 );
fToBeCompressedCapacity = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
fToBeCompressedCapacity = CompressInterface::UNCOMPRESSED_INBUF_LEN;
}
if ((fNumBytes + writeSize + fillUpWEmptiesWriteSize) <= fToBeCompressedCapacity)
@ -316,12 +320,12 @@ int ColumnBufferCompressed::writeToFile(int startOffset, int writeSize,
// Start over again loading a new to-be-compressed buffer
BlockOp::setEmptyBuf( fToBeCompressedBuffer,
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN,
CompressInterface::UNCOMPRESSED_INBUF_LEN,
fColInfo->column.emptyVal,
fColInfo->column.width );
fToBeCompressedCapacity =
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
CompressInterface::UNCOMPRESSED_INBUF_LEN;
bufOffset = fToBeCompressedBuffer;
fNumBytes = 0;
@ -377,21 +381,31 @@ int ColumnBufferCompressed::writeToFile(int startOffset, int writeSize,
//------------------------------------------------------------------------------
int ColumnBufferCompressed::compressAndFlush( bool bFinishingFile )
{
const int OUTPUT_BUFFER_SIZE = IDBCompressInterface::maxCompressedSize(fToBeCompressedCapacity) +
fUserPaddingBytes;
auto compressor = compress::getCompressorByType(
fCompressorPool, fColInfo->column.compressionType);
if (!compressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
const size_t OUTPUT_BUFFER_SIZE =
compressor->maxCompressedSize(fToBeCompressedCapacity) +
fUserPaddingBytes +
// Padded len = len + COMPRESSED_SIZE_INCREMENT_CHUNK - (len %
// COMPRESSED_SIZE_INCREMENT_CHUNK) + usePadding
compress::CompressInterface::COMPRESSED_CHUNK_INCREMENT_SIZE;
unsigned char* compressedOutBuf = new unsigned char[ OUTPUT_BUFFER_SIZE ];
boost::scoped_array<unsigned char> compressedOutBufPtr(compressedOutBuf);
unsigned int outputLen = OUTPUT_BUFFER_SIZE;
size_t outputLen = OUTPUT_BUFFER_SIZE;
#ifdef PROFILE
Stats::startParseEvent(WE_STATS_COMPRESS_COL_COMPRESS);
#endif
int rc = fCompressor->compressBlock(
reinterpret_cast<char*>(fToBeCompressedBuffer),
fToBeCompressedCapacity,
compressedOutBuf,
outputLen );
int rc = compressor->compressBlock(
reinterpret_cast<char*>(fToBeCompressedBuffer),
fToBeCompressedCapacity, compressedOutBuf, outputLen);
if (rc != 0)
{
@ -399,7 +413,7 @@ int ColumnBufferCompressed::compressAndFlush( bool bFinishingFile )
}
// Round up the compressed chunk size
rc = fCompressor->padCompressedChunks( compressedOutBuf,
rc = compressor->padCompressedChunks( compressedOutBuf,
outputLen, OUTPUT_BUFFER_SIZE );
if (rc != 0)
@ -581,26 +595,24 @@ int ColumnBufferCompressed::finishFile(bool bTruncFile)
int ColumnBufferCompressed::saveCompressionHeaders( )
{
// Construct the header records
char hdrBuf[IDBCompressInterface::HDR_BUF_LEN * 2];
char hdrBuf[CompressInterface::HDR_BUF_LEN * 2];
RETURN_ON_ERROR(fColInfo->colOp->readHeaders(fFile, hdrBuf));
BRM::LBID_t lbid = fCompressor->getLBIDByIndex(hdrBuf, 0);
fCompressor->initHdr(hdrBuf, fColInfo->column.width,
fColInfo->column.dataType,
fColInfo->column.compressionType);
fCompressor->setBlockCount(hdrBuf,
(fColInfo->getFileSize() / BYTE_PER_BLOCK) );
BRM::LBID_t lbid = compress::CompressInterface::getLBIDByIndex(hdrBuf, 0);
compress::CompressInterface::initHdr(hdrBuf, fColInfo->column.width, fColInfo->column.dataType,
fColInfo->column.compressionType);
compress::CompressInterface::setBlockCount(hdrBuf, (fColInfo->getFileSize() / BYTE_PER_BLOCK));
// If lbid written in the header is not 0 and not equal to `lastupdatedlbid` - we are running
// for the next extent for column segment file.
const auto lastUpdatedLbid = fColInfo->getLastUpdatedLBID();
if (lbid && lastUpdatedLbid != lbid)
{
// Write back lbid, after header initialization.
fCompressor->setLBIDByIndex(hdrBuf, lbid, 0);
fCompressor->setLBIDByIndex(hdrBuf, lastUpdatedLbid, 1);
compress::CompressInterface::setLBIDByIndex(hdrBuf, lbid, 0);
compress::CompressInterface::setLBIDByIndex(hdrBuf, lastUpdatedLbid, 1);
}
else
fCompressor->setLBIDByIndex(hdrBuf, fColInfo->getLastUpdatedLBID(), 0);
compress::CompressInterface::setLBIDByIndex(hdrBuf, fColInfo->getLastUpdatedLBID(), 0);
std::vector<uint64_t> ptrs;
@ -611,7 +623,7 @@ int ColumnBufferCompressed::saveCompressionHeaders( )
unsigned lastIdx = fChunkPtrs.size() - 1;
ptrs.push_back( fChunkPtrs[lastIdx].first + fChunkPtrs[lastIdx].second );
fCompressor->storePtrs( ptrs, hdrBuf );
compress::CompressInterface::storePtrs(ptrs, hdrBuf);
// Write out the header records
//char resp;
@ -641,9 +653,9 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
if (!fToBeCompressedBuffer)
{
fToBeCompressedBuffer =
new unsigned char[IDBCompressInterface::UNCOMPRESSED_INBUF_LEN];
new unsigned char[CompressInterface::UNCOMPRESSED_INBUF_LEN];
BlockOp::setEmptyBuf( fToBeCompressedBuffer,
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN,
CompressInterface::UNCOMPRESSED_INBUF_LEN,
fColInfo->column.emptyVal,
fColInfo->column.width );
bNewBuffer = true;
@ -656,12 +668,19 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
unsigned int blockOffsetWithinChunk = 0;
bool bSkipStartingBlks = false;
auto compressor = compress::getCompressorByType(
fCompressorPool, fColInfo->column.compressionType);
if (!compressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
if (fPreLoadHWMChunk)
{
if (fChunkPtrs.size() > 0)
{
fCompressor->locateBlock(fStartingHwm,
chunkIndex, blockOffsetWithinChunk);
compressor->locateBlock(fStartingHwm, chunkIndex,
blockOffsetWithinChunk);
if (chunkIndex < fChunkPtrs.size())
startFileOffset = fChunkPtrs[chunkIndex].first;
@ -718,8 +737,8 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
}
// Uncompress the chunk into our 4MB buffer
unsigned int outLen = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
int rc = fCompressor->uncompressBlock(
size_t outLen = CompressInterface::UNCOMPRESSED_INBUF_LEN;
int rc = compressor->uncompressBlock(
compressedOutBuf,
fChunkPtrs[chunkIndex].second,
fToBeCompressedBuffer,
@ -758,7 +777,7 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
if (!bNewBuffer)
{
BlockOp::setEmptyBuf( fToBeCompressedBuffer,
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN,
CompressInterface::UNCOMPRESSED_INBUF_LEN,
fColInfo->column.emptyVal,
fColInfo->column.width );
}
@ -775,10 +794,10 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
fLog->logMsg( oss.str(), MSGLVL_INFO2 );
}
fToBeCompressedCapacity = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
fToBeCompressedCapacity = CompressInterface::UNCOMPRESSED_INBUF_LEN;
// Set file offset to start after last current chunk
startFileOffset = IDBCompressInterface::HDR_BUF_LEN * 2;
startFileOffset = CompressInterface::HDR_BUF_LEN * 2;
if (fChunkPtrs.size() > 0)
startFileOffset = fChunkPtrs[ fChunkPtrs.size() - 1 ].first +
@ -796,5 +815,4 @@ int ColumnBufferCompressed::initToBeCompressedBuffer(long long& startFileOffset)
return NO_ERROR;
}
}

3
writeengine/bulk/we_colbufcompressed.h
View File

@ -107,8 +107,7 @@ private:
// should always be 4MB, unless
// working with abbrev extent.
size_t fNumBytes; // num Bytes in comp buffer
compress::IDBCompressInterface*
fCompressor; // data compression object
compress::CompressorPool fCompressorPool; // data compression object pool
compress::CompChunkPtrList
fChunkPtrs; // col file header information
bool fPreLoadHWMChunk; // preload 1st HWM chunk only

7
writeengine/bulk/we_columninfo.cpp
View File

@ -450,7 +450,7 @@ int ColumnInfo::createDelayedFileIfNeeded( const std::string& tableName )
if (column.dctnry.fCompressionType != 0)
{
DctnryCompress1* tempD1;
tempD1 = new DctnryCompress1;
tempD1 = new DctnryCompress1(column.dctnry.fCompressionType);
tempD1->setMaxActiveChunkNum(1);
tempD1->setBulkFlag(true);
tempD = tempD1;
@ -668,7 +668,7 @@ int ColumnInfo::extendColumnNewExtent(
uint16_t segmentNew = 0;
BRM::LBID_t startLbid;
char hdr[ compress::IDBCompressInterface::HDR_BUF_LEN * 2 ];
char hdr[ compress::CompressInterface::HDR_BUF_LEN * 2 ];
// Extend the column by adding an extent to the next
// DBRoot, partition, and segment file in the rotation
@ -1684,7 +1684,8 @@ int ColumnInfo::openDctnryStore( bool bMustExist )
if ( column.dctnry.fCompressionType != 0)
{
DctnryCompress1* dctnryCompress1 = new DctnryCompress1;
DctnryCompress1* dctnryCompress1 =
new DctnryCompress1(column.dctnry.fCompressionType);
dctnryCompress1->setMaxActiveChunkNum(1);
dctnryCompress1->setBulkFlag(true);
fStore = dctnryCompress1;

26
writeengine/bulk/we_columninfocompressed.cpp
View File

@ -108,7 +108,7 @@ int ColumnInfoCompressed::closeColumnFile(bool bCompletingExtent, bool bAbort)
//------------------------------------------------------------------------------
int ColumnInfoCompressed::setupInitialColumnFile( HWM oldHwm, HWM hwm )
{
char hdr[ compress::IDBCompressInterface::HDR_BUF_LEN * 2 ];
char hdr[ compress::CompressInterface::HDR_BUF_LEN * 2 ];
RETURN_ON_ERROR( colOp->readHeaders(curCol.dataFile.pFile, hdr) );
// Initialize the output buffer manager for the column.
@ -129,10 +129,9 @@ int ColumnInfoCompressed::setupInitialColumnFile( HWM oldHwm, HWM hwm )
fColBufferMgr = mgr;
IDBCompressInterface compressor;
int abbrevFlag =
( compressor.getBlockCount(hdr) ==
uint64_t(INITIAL_EXTENT_ROWS_TO_DISK * column.width / BYTE_PER_BLOCK) );
int abbrevFlag = (compress::CompressInterface::getBlockCount(hdr) ==
uint64_t(INITIAL_EXTENT_ROWS_TO_DISK * column.width /
BYTE_PER_BLOCK));
setFileSize( hwm, abbrevFlag );
// See if dealing with abbreviated extent that will need expanding.
@ -324,9 +323,9 @@ int ColumnInfoCompressed::truncateDctnryStore(
return rc;
}
char controlHdr[ IDBCompressInterface::HDR_BUF_LEN ];
char controlHdr[ CompressInterface::HDR_BUF_LEN ];
rc = fTruncateDctnryFileOp.readFile( dFile,
(unsigned char*)controlHdr, IDBCompressInterface::HDR_BUF_LEN);
(unsigned char*)controlHdr, CompressInterface::HDR_BUF_LEN);
if (rc != NO_ERROR)
{
@ -345,8 +344,7 @@ int ColumnInfoCompressed::truncateDctnryStore(
return rc;
}
IDBCompressInterface compressor;
int rc1 = compressor.verifyHdr( controlHdr );
int rc1 = compress::CompressInterface::verifyHdr(controlHdr);
if (rc1 != 0)
{
@ -372,7 +370,8 @@ int ColumnInfoCompressed::truncateDctnryStore(
// actually grow the file (something we don't want to do), because we have
// not yet reserved a full extent (on disk) for this dictionary store file.
const int PSEUDO_COL_WIDTH = 8;
uint64_t numBlocks = compressor.getBlockCount( controlHdr );
uint64_t numBlocks =
compress::CompressInterface::getBlockCount(controlHdr);
if ( numBlocks == uint64_t
(INITIAL_EXTENT_ROWS_TO_DISK * PSEUDO_COL_WIDTH / BYTE_PER_BLOCK) )
@ -390,8 +389,8 @@ int ColumnInfoCompressed::truncateDctnryStore(
return NO_ERROR;
}
uint64_t hdrSize = compressor.getHdrSize(controlHdr);
uint64_t ptrHdrSize = hdrSize - IDBCompressInterface::HDR_BUF_LEN;
uint64_t hdrSize = compress::CompressInterface::getHdrSize(controlHdr);
uint64_t ptrHdrSize = hdrSize - CompressInterface::HDR_BUF_LEN;
char* pointerHdr = new char[ptrHdrSize];
rc = fTruncateDctnryFileOp.readFile(dFile,
@ -416,7 +415,8 @@ int ColumnInfoCompressed::truncateDctnryStore(
}
CompChunkPtrList chunkPtrs;
rc1 = compressor.getPtrList( pointerHdr, ptrHdrSize, chunkPtrs );
rc1 = compress::CompressInterface::getPtrList(pointerHdr, ptrHdrSize,
chunkPtrs);
delete[] pointerHdr;
if (rc1 != 0)

15
writeengine/server/we_getfilesizes.cpp
View File

@ -96,7 +96,7 @@ size_t readFillBuffer(
return totalBytesRead;
}
off64_t getCompressedDataSize(string& fileName)
static off64_t getCompressedDataSize(string& fileName)
{
off64_t dataSize = 0;
IDBDataFile* pFile = 0;
@ -119,21 +119,21 @@ off64_t getCompressedDataSize(string& fileName)
throw std::runtime_error(oss.str());
}
IDBCompressInterface decompressor;
//--------------------------------------------------------------------------
// Read headers and extract compression pointers
//--------------------------------------------------------------------------
char hdr1[IDBCompressInterface::HDR_BUF_LEN];
nBytes = readFillBuffer( pFile, hdr1, IDBCompressInterface::HDR_BUF_LEN);
char hdr1[CompressInterface::HDR_BUF_LEN];
nBytes = readFillBuffer( pFile, hdr1, CompressInterface::HDR_BUF_LEN);
if ( nBytes != IDBCompressInterface::HDR_BUF_LEN )
if ( nBytes != CompressInterface::HDR_BUF_LEN )
{
std::ostringstream oss;
oss << "Error reading first header from file " << fileName;
throw std::runtime_error(oss.str());
}
int64_t ptrSecSize = decompressor.getHdrSize(hdr1) - IDBCompressInterface::HDR_BUF_LEN;
int64_t ptrSecSize = compress::CompressInterface::getHdrSize(hdr1) -
CompressInterface::HDR_BUF_LEN;
char* hdr2 = new char[ptrSecSize];
nBytes = readFillBuffer( pFile, hdr2, ptrSecSize);
@ -145,7 +145,8 @@ off64_t getCompressedDataSize(string& fileName)
}
CompChunkPtrList chunkPtrs;
int rc = decompressor.getPtrList(hdr2, ptrSecSize, chunkPtrs);
int rc =
compress::CompressInterface::getPtrList(hdr2, ptrSecSize, chunkPtrs);
delete[] hdr2;
if (rc != 0)

81
writeengine/shared/we_bulkrollbackfilecompressed.cpp
View File

@ -51,6 +51,7 @@ namespace WriteEngine
BulkRollbackFileCompressed::BulkRollbackFileCompressed(BulkRollbackMgr* mgr) :
BulkRollbackFile(mgr)
{
compress::initializeCompressorPool(fCompressorPool);
}
//------------------------------------------------------------------------------
@ -104,7 +105,7 @@ void BulkRollbackFileCompressed::truncateSegmentFile(
}
// Read and parse the header pointers
char hdrs[ IDBCompressInterface::HDR_BUF_LEN * 2 ];;
char hdrs[ CompressInterface::HDR_BUF_LEN * 2 ];;
CompChunkPtrList chunkPtrs;
std::string errMsg;
int rc = loadColumnHdrPtrs(pFile, hdrs, chunkPtrs, errMsg);
@ -127,7 +128,20 @@ void BulkRollbackFileCompressed::truncateSegmentFile(
unsigned int blockOffset = fileSizeBlocks - 1;
unsigned int chunkIndex = 0;
unsigned int blkOffsetInChunk = 0;
fCompressor.locateBlock( blockOffset, chunkIndex, blkOffsetInChunk );
auto fCompressor = compress::getCompressorByType(
fCompressorPool,
compress::CompressInterface::getCompressionType(hdrs));
if (!fCompressor)
{
std::ostringstream oss;
oss << "Error, wrong compression type for segment file"
<< ": OID-" << columnOID << "; DbRoot-" << dbRoot << "; partition-"
<< partNum << "; segment-" << segNum << ";";
throw WeException(oss.str(), ERR_COMP_WRONG_COMP_TYPE);
}
fCompressor->locateBlock(blockOffset, chunkIndex, blkOffsetInChunk);
// Truncate the extra extents that are to be aborted
if (chunkIndex < chunkPtrs.size())
@ -145,7 +159,7 @@ void BulkRollbackFileCompressed::truncateSegmentFile(
logging::M0075, columnOID, msgText2.str() );
// Drop off any trailing pointers (that point beyond the last block)
fCompressor.setBlockCount( hdrs, fileSizeBlocks );
compress::CompressInterface::setBlockCount(hdrs, fileSizeBlocks);
std::vector<uint64_t> ptrs;
for (unsigned i = 0; i <= chunkIndex; i++)
@ -155,7 +169,7 @@ void BulkRollbackFileCompressed::truncateSegmentFile(
ptrs.push_back( chunkPtrs[chunkIndex].first +
chunkPtrs[chunkIndex].second );
fCompressor.storePtrs( ptrs, hdrs );
compress::CompressInterface::storePtrs(ptrs, hdrs);
rc = fDbFile.writeHeaders( pFile, hdrs );
@ -252,7 +266,7 @@ void BulkRollbackFileCompressed::reInitTruncColumnExtent(
}
// Read and parse the header pointers
char hdrs[ IDBCompressInterface::HDR_BUF_LEN * 2 ];
char hdrs[ CompressInterface::HDR_BUF_LEN * 2 ];
CompChunkPtrList chunkPtrs;
std::string errMsg;
int rc = loadColumnHdrPtrs(pFile, hdrs, chunkPtrs, errMsg);
@ -275,7 +289,20 @@ void BulkRollbackFileCompressed::reInitTruncColumnExtent(
unsigned int blockOffset = startOffsetBlk - 1;
unsigned int chunkIndex = 0;
unsigned int blkOffsetInChunk = 0;
fCompressor.locateBlock( blockOffset, chunkIndex, blkOffsetInChunk );
auto fCompressor = compress::getCompressorByType(
fCompressorPool,
compress::CompressInterface::getCompressionType(hdrs));
if (!fCompressor)
{
std::ostringstream oss;
oss << "Error, wrong compression type for segment file"
<< ": OID-" << columnOID << "; DbRoot-" << dbRoot << "; partition-"
<< partNum << "; segment-" << segNum << ";";
throw WeException(oss.str(), ERR_COMP_WRONG_COMP_TYPE);
}
fCompressor->locateBlock(blockOffset, chunkIndex, blkOffsetInChunk);
if (chunkIndex < chunkPtrs.size())
{
@ -401,7 +428,8 @@ void BulkRollbackFileCompressed::reInitTruncColumnExtent(
// Watch for the special case where we are restoring a db file as an
// empty file (chunkindex=0 and restoredChunkLen=0); in this case we
// just restore the first pointer (set to 8192).
fCompressor.setBlockCount( hdrs, (startOffsetBlk + nBlocks) );
compress::CompressInterface::setBlockCount(hdrs,
(startOffsetBlk + nBlocks));
std::vector<uint64_t> newPtrs;
if ((chunkIndex > 0) || (restoredChunkLen > 0))
@ -413,7 +441,7 @@ void BulkRollbackFileCompressed::reInitTruncColumnExtent(
}
newPtrs.push_back( chunkPtrs[chunkIndex].first + restoredChunkLen );
fCompressor.storePtrs( newPtrs, hdrs );
compress::CompressInterface::storePtrs(newPtrs, hdrs);
rc = fDbFile.writeHeaders( pFile, hdrs );
@ -482,7 +510,7 @@ int BulkRollbackFileCompressed::loadColumnHdrPtrs(
}
// Parse the header pointers
int rc1 = fCompressor.getPtrList( hdrs, chunkPtrs );
int rc1 = compress::CompressInterface::getPtrList(hdrs, chunkPtrs);
if (rc1 != 0)
{
@ -548,7 +576,7 @@ void BulkRollbackFileCompressed::reInitTruncDctnryExtent(
throw WeException( oss.str(), ERR_FILE_OPEN );
}
char controlHdr[ IDBCompressInterface::HDR_BUF_LEN ];
char controlHdr[ CompressInterface::HDR_BUF_LEN ];
CompChunkPtrList chunkPtrs;
uint64_t ptrHdrSize;
std::string errMsg;
@ -572,7 +600,20 @@ void BulkRollbackFileCompressed::reInitTruncDctnryExtent(
unsigned int blockOffset = startOffsetBlk - 1;
unsigned int chunkIndex = 0;
unsigned int blkOffsetInChunk = 0;
fCompressor.locateBlock( blockOffset, chunkIndex, blkOffsetInChunk );
auto fCompressor = compress::getCompressorByType(
fCompressorPool,
compress::CompressInterface::getCompressionType(controlHdr));
if (!fCompressor)
{
std::ostringstream oss;
oss << "Error, wrong compression type for segment file"
<< ": OID-" << dStoreOID << "; DbRoot-" << dbRoot << "; partition-"
<< partNum << "; segment-" << segNum << ";";
throw WeException(oss.str(), ERR_COMP_WRONG_COMP_TYPE);
}
fCompressor->locateBlock(blockOffset, chunkIndex, blkOffsetInChunk);
if (chunkIndex < chunkPtrs.size())
{
@ -686,7 +727,8 @@ void BulkRollbackFileCompressed::reInitTruncDctnryExtent(
// Watch for the special case where we are restoring a db file as an
// empty file (chunkindex=0 and restoredChunkLen=0); in this case we
// just restore the first pointer (set to 8192).
fCompressor.setBlockCount( controlHdr, (startOffsetBlk + nBlocks) );
compress::CompressInterface::setBlockCount(controlHdr,
(startOffsetBlk + nBlocks));
std::vector<uint64_t> newPtrs;
if ((chunkIndex > 0) || (restoredChunkLen > 0))
@ -699,7 +741,8 @@ void BulkRollbackFileCompressed::reInitTruncDctnryExtent(
newPtrs.push_back( chunkPtrs[chunkIndex].first + restoredChunkLen );
char* pointerHdr = new char[ptrHdrSize];
fCompressor.storePtrs( newPtrs, pointerHdr, ptrHdrSize );
compress::CompressInterface::storePtrs(newPtrs, pointerHdr,
ptrHdrSize);
rc = fDbFile.writeHeaders( pFile, controlHdr, pointerHdr, ptrHdrSize );
delete[] pointerHdr;
@ -759,7 +802,7 @@ int BulkRollbackFileCompressed::loadDctnryHdrPtrs(
std::string& errMsg) const
{
int rc = fDbFile.readFile(
pFile, (unsigned char*)controlHdr, IDBCompressInterface::HDR_BUF_LEN);
pFile, (unsigned char*)controlHdr, CompressInterface::HDR_BUF_LEN);
if (rc != NO_ERROR)
{
@ -771,7 +814,7 @@ int BulkRollbackFileCompressed::loadDctnryHdrPtrs(
return rc;
}
int rc1 = fCompressor.verifyHdr( controlHdr );
int rc1 = compress::CompressInterface::verifyHdr(controlHdr);
if (rc1 != 0)
{
@ -786,8 +829,8 @@ int BulkRollbackFileCompressed::loadDctnryHdrPtrs(
return rc;
}
uint64_t hdrSize = fCompressor.getHdrSize(controlHdr);
ptrHdrSize = hdrSize - IDBCompressInterface::HDR_BUF_LEN;
uint64_t hdrSize = compress::CompressInterface::getHdrSize(controlHdr);
ptrHdrSize = hdrSize - CompressInterface::HDR_BUF_LEN;
char* pointerHdr = new char[ptrHdrSize];
rc = fDbFile.readFile(pFile, (unsigned char*)pointerHdr, ptrHdrSize);
@ -804,7 +847,8 @@ int BulkRollbackFileCompressed::loadDctnryHdrPtrs(
}
// Parse the header pointers
rc1 = fCompressor.getPtrList( pointerHdr, ptrHdrSize, chunkPtrs );
rc1 = compress::CompressInterface::getPtrList(pointerHdr, ptrHdrSize,
chunkPtrs);
delete[] pointerHdr;
if (rc1 != 0)
@ -1033,5 +1077,4 @@ size_t BulkRollbackFileCompressed::readFillBuffer(
return totalBytesRead;
}
} //end of namespace

3
writeengine/shared/we_bulkrollbackfilecompressed.h
View File

@ -28,6 +28,7 @@
#include <cstdio>
#include <cstring>
#include <unordered_map>
#include "we_define.h"
#include "we_type.h"
@ -148,7 +149,7 @@ private:
uint64_t& ptrHdrSize,
std::string& errMsg ) const;
compress::IDBCompressInterface fCompressor;
compress::CompressorPool fCompressorPool;
};
} //end of namespace

164
writeengine/shared/we_chunkmanager.cpp
View File

@ -67,8 +67,6 @@ namespace WriteEngine
extern int NUM_BLOCKS_PER_INITIAL_EXTENT; // defined in we_dctnry.cpp
extern WErrorCodes ec; // defined in we_log.cpp
const int COMPRESSED_CHUNK_SIZE = compress::IDBCompressInterface::maxCompressedSize(UNCOMPRESSED_CHUNK_SIZE) + 64 + 3 + 8 * 1024;
//------------------------------------------------------------------------------
// Search for the specified chunk in fChunkList.
//------------------------------------------------------------------------------
@ -91,18 +89,24 @@ ChunkData* CompFileData::findChunk(int64_t id) const
//------------------------------------------------------------------------------
// ChunkManager constructor
//------------------------------------------------------------------------------
ChunkManager::ChunkManager() : fMaxActiveChunkNum(100), fLenCompressed(0), fIsBulkLoad(false),
fDropFdCache(false), fIsInsert(false), fIsHdfs(IDBPolicy::useHdfs()),
fFileOp(0), fSysLogger(NULL), fTransId(-1),
fLocalModuleId(Config::getLocalModuleID()),
fFs(fIsHdfs ?
IDBFileSystem::getFs(IDBDataFile::HDFS) :
IDBPolicy::useCloud() ?
IDBFileSystem::getFs(IDBDataFile::CLOUD) :
IDBFileSystem::getFs(IDBDataFile::BUFFERED))
ChunkManager::ChunkManager()
: fMaxActiveChunkNum(100), fLenCompressed(0), fIsBulkLoad(false),
fDropFdCache(false), fIsInsert(false), fIsHdfs(IDBPolicy::useHdfs()),
fFileOp(0), fSysLogger(NULL), fTransId(-1),
fLocalModuleId(Config::getLocalModuleID()),
fFs(fIsHdfs ? IDBFileSystem::getFs(IDBDataFile::HDFS)
: IDBPolicy::useCloud()
? IDBFileSystem::getFs(IDBDataFile::CLOUD)
: IDBFileSystem::getFs(IDBDataFile::BUFFERED))
{
fUserPaddings = Config::getNumCompressedPadBlks() * BYTE_PER_BLOCK;
fCompressor.numUserPaddingBytes(fUserPaddings);
compress::initializeCompressorPool(fCompressorPool, fUserPaddings);
COMPRESSED_CHUNK_SIZE =
compress::CompressInterface::getMaxCompressedSizeGeneric(
UNCOMPRESSED_CHUNK_SIZE) +
64 + 3 + 8 * 1024;
fMaxCompressedBufSize = COMPRESSED_CHUNK_SIZE + fUserPaddings;
fBufCompressed = new char[fMaxCompressedBufSize];
fSysLogger = new logging::Logger(SUBSYSTEM_ID_WE);
@ -383,16 +387,22 @@ CompFileData* ChunkManager::getFileData(const FID& fid,
}
// make sure the header is valid
if (fCompressor.verifyHdr(fileData->fFileHeader.fControlData) != 0)
if (compress::CompressInterface::verifyHdr(fileData->fFileHeader.fControlData) != 0)
{
WE_COMP_DBG(cout << "Invalid header." << endl;)
delete fileData;
return NULL;
}
int headerSize = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
// Save segment file compression type.
uint32_t compressionType = compress::CompressInterface::getCompressionType(
fileData->fFileHeader.fControlData);
fileData->fCompressionType = compressionType;
if (ptrSecSize > COMPRESSED_FILE_HEADER_UNIT)
{
// >8K header, dictionary width > 128
@ -462,11 +472,12 @@ IDBDataFile* ChunkManager::createDctnryFile(const FID& fid,
// Dictionary store extent width == 0. See more details in function
// `createDictStoreExtent`.
fCompressor.initHdr(fileData->fFileHeader.fControlData,
fileData->fFileHeader.fPtrSection,
/*colWidth=*/0, fileData->fColDataType,
fFileOp->compressionType(), hdrSize);
fCompressor.setLBIDByIndex(fileData->fFileHeader.fControlData, lbid, 0);
compress::CompressInterface::initHdr(
fileData->fFileHeader.fControlData, fileData->fFileHeader.fPtrSection,
/*colWidth=*/0, fileData->fColDataType, fFileOp->compressionType(), hdrSize);
compress::CompressInterface::setLBIDByIndex(fileData->fFileHeader.fControlData, lbid, 0);
// Save compression type.
fileData->fCompressionType = fFileOp->compressionType();
if (writeHeader(fileData, __LINE__) != NO_ERROR)
{
@ -771,9 +782,16 @@ int ChunkManager::fetchChunkFromFile(IDBDataFile* pFile, int64_t id, ChunkData*&
}
// uncompress the read in buffer
unsigned int dataLen = sizeof(chunkData->fBufUnCompressed);
size_t dataLen = sizeof(chunkData->fBufUnCompressed);
if (fCompressor.uncompressBlock((char*)fBufCompressed, chunkSize,
auto fCompressor = compress::getCompressorByType(
fCompressorPool, fileData->fCompressionType);
if (!fCompressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
if (fCompressor->uncompressBlock((char*)fBufCompressed, chunkSize,
(unsigned char*)chunkData->fBufUnCompressed, dataLen) != 0)
{
if (fIsFix)
@ -784,7 +802,7 @@ int ChunkManager::fetchChunkFromFile(IDBDataFile* pFile, int64_t id, ChunkData*&
{
char* hdr = fileData->fFileHeader.fControlData;
if (fCompressor.getBlockCount(hdr) < 512)
if (compress::CompressInterface::getBlockCount(hdr) < 512)
blocks = 256;
}
@ -820,7 +838,8 @@ int ChunkManager::fetchChunkFromFile(IDBDataFile* pFile, int64_t id, ChunkData*&
{
if (id == 0 && ptrs[id] == 0) // if the 1st ptr is not set for new extent
{
ptrs[0] = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
ptrs[0] = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
}
// load the uncompressed buffer with empty values.
@ -907,10 +926,17 @@ int ChunkManager::writeChunkToFile(CompFileData* fileData, ChunkData* chunkData)
// compress the chunk before writing it to file
fLenCompressed = fMaxCompressedBufSize;
if (fCompressor.compressBlock((char*)chunkData->fBufUnCompressed,
chunkData->fLenUnCompressed,
(unsigned char*)fBufCompressed,
fLenCompressed) != 0)
auto fCompressor = compress::getCompressorByType(
fCompressorPool, fileData->fCompressionType);
if (!fCompressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
if (fCompressor->compressBlock((char*) chunkData->fBufUnCompressed,
chunkData->fLenUnCompressed,
(unsigned char*) fBufCompressed,
fLenCompressed) != 0)
{
logMessage(ERR_COMP_COMPRESS, logging::LOG_TYPE_ERROR, __LINE__);
return ERR_COMP_COMPRESS;
@ -941,7 +967,8 @@ int ChunkManager::writeChunkToFile(CompFileData* fileData, ChunkData* chunkData)
// [chunkId+0] is the start offset of current chunk.
// [chunkId+1] is the start offset of next chunk, the offset diff is current chunk size.
// [chunkId+2] is 0 or not indicates if the next chunk exists.
int headerSize = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
int64_t usablePtrIds = (ptrSecSize / sizeof(uint64_t)) - 2;
@ -968,7 +995,7 @@ int ChunkManager::writeChunkToFile(CompFileData* fileData, ChunkData* chunkData)
else if (lastChunk)
{
// add padding space if the chunk is written first time
if (fCompressor.padCompressedChunks(
if (fCompressor->padCompressedChunks(
(unsigned char*)fBufCompressed, fLenCompressed, fMaxCompressedBufSize) != 0)
{
WE_COMP_DBG(cout << "Last chunk:" << chunkId << ", padding failed." << endl;)
@ -1272,7 +1299,8 @@ int ChunkManager::closeFile(CompFileData* fileData)
int ChunkManager::writeHeader(CompFileData* fileData, int ln)
{
int rc = NO_ERROR;
int headerSize = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
if (!fIsHdfs && !fIsBulkLoad)
@ -1422,8 +1450,10 @@ int ChunkManager::updateColumnExtent(IDBDataFile* pFile, int addBlockCount, int6
int rc = NO_ERROR;
char* hdr = pFileData->fFileHeader.fControlData;
fCompressor.setBlockCount(hdr, fCompressor.getBlockCount(hdr) + addBlockCount);
fCompressor.setLBIDByIndex(hdr, lbid, 1);
compress::CompressInterface::setBlockCount(
hdr, compress::CompressInterface::getBlockCount(hdr) + addBlockCount);
compress::CompressInterface::setLBIDByIndex(hdr, lbid, 1);
ChunkData* chunkData = (pFileData)->findChunk(0);
if (chunkData != NULL)
@ -1475,7 +1505,7 @@ int ChunkManager::updateDctnryExtent(IDBDataFile* pFile, int addBlockCount,
char* hdr = i->second->fFileHeader.fControlData;
char* uncompressedBuf = chunkData->fBufUnCompressed;
int currentBlockCount = fCompressor.getBlockCount(hdr);
int currentBlockCount = compress::CompressInterface::getBlockCount(hdr);
// Bug 3203, write out the compressed initial extent.
if (currentBlockCount == 0)
@ -1511,13 +1541,15 @@ int ChunkManager::updateDctnryExtent(IDBDataFile* pFile, int addBlockCount,
}
if (rc == NO_ERROR)
fCompressor.setBlockCount(hdr, fCompressor.getBlockCount(hdr) + addBlockCount);
compress::CompressInterface::setBlockCount(
hdr,
compress::CompressInterface::getBlockCount(hdr) + addBlockCount);
if (currentBlockCount)
{
// Append to the end.
uint64_t lbidCount = fCompressor.getLBIDCount(hdr);
fCompressor.setLBIDByIndex(hdr, lbid, lbidCount);
uint64_t lbidCount = compress::CompressInterface::getLBIDCount(hdr);
compress::CompressInterface::setLBIDByIndex(hdr, lbid, lbidCount);
}
return rc;
}
@ -1684,7 +1716,8 @@ int ChunkManager::getBlockCount(IDBDataFile* pFile)
map<IDBDataFile*, CompFileData*>::iterator fpIt = fFilePtrMap.find(pFile);
idbassert(fpIt != fFilePtrMap.end());
return fCompressor.getBlockCount(fpIt->second->fFileHeader.fControlData);
return compress::CompressInterface::getBlockCount(
fpIt->second->fFileHeader.fControlData);
}
//------------------------------------------------------------------------------
@ -1758,11 +1791,13 @@ int ChunkManager::reallocateChunks(CompFileData* fileData)
origFilePtr->flush();
// back out the current pointers
int headerSize = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
compress::CompChunkPtrList origPtrs;
if (fCompressor.getPtrList(fileData->fFileHeader.fPtrSection, ptrSecSize, origPtrs) != 0)
if (compress::CompressInterface::getPtrList(
fileData->fFileHeader.fPtrSection, ptrSecSize, origPtrs) != 0)
{
ostringstream oss;
oss << "Chunk shifting failed, file:" << origFileName << " -- invalid header.";
@ -1876,7 +1911,14 @@ int ChunkManager::reallocateChunks(CompFileData* fileData)
ChunkData* chunkData = chunksTouched[k];
fLenCompressed = fMaxCompressedBufSize;
if ((rc = fCompressor.compressBlock((char*)chunkData->fBufUnCompressed,
auto fCompressor = compress::getCompressorByType(
fCompressorPool, fileData->fCompressionType);
if (!fCompressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
if ((rc = fCompressor->compressBlock((char*)chunkData->fBufUnCompressed,
chunkData->fLenUnCompressed,
(unsigned char*)fBufCompressed,
fLenCompressed)) != 0)
@ -1894,7 +1936,7 @@ int ChunkManager::reallocateChunks(CompFileData* fileData)
<< fLenCompressed;)
// shifting chunk, add padding space
if ((rc = fCompressor.padCompressedChunks(
if ((rc = fCompressor->padCompressedChunks(
(unsigned char*)fBufCompressed, fLenCompressed, fMaxCompressedBufSize)) != 0)
{
WE_COMP_DBG(cout << ", but padding failed." << endl;)
@ -2245,7 +2287,8 @@ int ChunkManager::verifyChunksAfterRealloc(CompFileData* fileData)
}
// make sure the header is valid
if ((rc = fCompressor.verifyHdr(fileData->fFileHeader.fControlData)) != 0)
if ((rc = compress::CompressInterface::verifyHdr(
fileData->fFileHeader.fControlData)) != 0)
{
ostringstream oss;
oss << "Invalid header in new " << fileData->fFileName << ", roll back";
@ -2254,7 +2297,8 @@ int ChunkManager::verifyChunksAfterRealloc(CompFileData* fileData)
return rc;
}
int headerSize = fCompressor.getHdrSize(fileData->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
fileData->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
// read in the pointer section in header
@ -2270,7 +2314,8 @@ int ChunkManager::verifyChunksAfterRealloc(CompFileData* fileData)
// get pointer list
compress::CompChunkPtrList ptrs;
if (fCompressor.getPtrList(fileData->fFileHeader.fPtrSection, ptrSecSize, ptrs) != 0)
if (compress::CompressInterface::getPtrList(
fileData->fFileHeader.fPtrSection, ptrSecSize, ptrs) != 0)
{
ostringstream oss;
oss << "Failed to parse pointer list from new " << fileData->fFileName << "@" << __LINE__;
@ -2282,6 +2327,13 @@ int ChunkManager::verifyChunksAfterRealloc(CompFileData* fileData)
ChunkData chunkData;
int numOfChunks = ptrs.size(); // number of chunks in the file
auto fCompressor = compress::getCompressorByType(
fCompressorPool, fileData->fCompressionType);
if (!fCompressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
for (int i = 0; i < numOfChunks && rc == NO_ERROR; i++)
{
unsigned int chunkSize = ptrs[i].second;
@ -2304,9 +2356,9 @@ int ChunkManager::verifyChunksAfterRealloc(CompFileData* fileData)
}
// uncompress the read in buffer
unsigned int dataLen = sizeof(chunkData.fBufUnCompressed);
size_t dataLen = sizeof(chunkData.fBufUnCompressed);
if (fCompressor.uncompressBlock((char*)fBufCompressed, chunkSize,
if (fCompressor->uncompressBlock((char*)fBufCompressed, chunkSize,
(unsigned char*)chunkData.fBufUnCompressed, dataLen) != 0)
{
ostringstream oss;
@ -2624,13 +2676,15 @@ int ChunkManager::checkFixLastDictChunk(const FID& fid,
if (mit != fFileMap.end())
{
int headerSize = fCompressor.getHdrSize(mit->second->fFileHeader.fControlData);
int headerSize = compress::CompressInterface::getHdrSize(
mit->second->fFileHeader.fControlData);
int ptrSecSize = headerSize - COMPRESSED_FILE_HEADER_UNIT;
// get pointer list
compress::CompChunkPtrList ptrs;
if (fCompressor.getPtrList(mit->second->fFileHeader.fPtrSection, ptrSecSize, ptrs) != 0)
if (compress::CompressInterface::getPtrList(
mit->second->fFileHeader.fPtrSection, ptrSecSize, ptrs) != 0)
{
ostringstream oss;
oss << "Failed to parse pointer list from new " << mit->second->fFileName << "@" << __LINE__;
@ -2662,9 +2716,16 @@ int ChunkManager::checkFixLastDictChunk(const FID& fid,
// uncompress the read in buffer
chunkData = new ChunkData(numOfChunks - 1);
unsigned int dataLen = sizeof(chunkData->fBufUnCompressed);
size_t dataLen = sizeof(chunkData->fBufUnCompressed);
if (fCompressor.uncompressBlock((char*)fBufCompressed, chunkSize,
auto fCompressor = compress::getCompressorByType(
fCompressorPool, mit->second->fCompressionType);
if (!fCompressor)
{
return ERR_COMP_WRONG_COMP_TYPE;
}
if (fCompressor->uncompressBlock((char*)fBufCompressed, chunkSize,
(unsigned char*)chunkData->fBufUnCompressed, dataLen) != 0)
{
mit->second->fChunkList.push_back(chunkData);
@ -2676,7 +2737,7 @@ int ChunkManager::checkFixLastDictChunk(const FID& fid,
{
char* hdr = mit->second->fFileHeader.fControlData;
if (fCompressor.getBlockCount(hdr) < 512)
if (compress::CompressInterface::getBlockCount(hdr) < 512)
blocks = 256;
}
@ -2693,7 +2754,6 @@ int ChunkManager::checkFixLastDictChunk(const FID& fid,
return rc;
}
}
// vim:ts=4 sw=4:

18
writeengine/shared/we_chunkmanager.h
View File

@ -64,8 +64,8 @@ namespace WriteEngine
// forward reference
class FileOp;
const int UNCOMPRESSED_CHUNK_SIZE = compress::IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
const int COMPRESSED_FILE_HEADER_UNIT = compress::IDBCompressInterface::HDR_BUF_LEN;
const int UNCOMPRESSED_CHUNK_SIZE = compress::CompressInterface::UNCOMPRESSED_INBUF_LEN;
const int COMPRESSED_FILE_HEADER_UNIT = compress::CompressInterface::HDR_BUF_LEN;
// assume UNCOMPRESSED_CHUNK_SIZE > 0xBFFF (49151), 8 * 1024 bytes padding
@ -136,7 +136,7 @@ class CompFileData
public:
CompFileData(const FileID& id, const FID& fid, const execplan::CalpontSystemCatalog::ColDataType colDataType, int colWidth) :
fFileID(id), fFid(fid), fColDataType(colDataType), fColWidth(colWidth), fDctnryCol(false),
fFilePtr(NULL), fIoBSize(0) {}
fFilePtr(NULL), fIoBSize(0), fCompressionType(1) {}
ChunkData* findChunk(int64_t cid) const;
@ -152,6 +152,7 @@ protected:
std::list<ChunkData*> fChunkList;
boost::scoped_array<char> fIoBuffer;
size_t fIoBSize;
uint32_t fCompressionType;
friend class ChunkManager;
};
@ -369,22 +370,23 @@ protected:
std::list<std::pair<FileID, ChunkData*> > fActiveChunks;
unsigned int fMaxActiveChunkNum; // max active chunks per file
char* fBufCompressed;
unsigned int fLenCompressed;
unsigned int fMaxCompressedBufSize;
unsigned int fUserPaddings;
size_t fLenCompressed;
size_t fMaxCompressedBufSize;
size_t fUserPaddings;
bool fIsBulkLoad;
bool fDropFdCache;
bool fIsInsert;
bool fIsHdfs;
FileOp* fFileOp;
compress::IDBCompressInterface fCompressor;
compress::CompressorPool fCompressorPool;
logging::Logger* fSysLogger;
TxnID fTransId;
int fLocalModuleId;
idbdatafile::IDBFileSystem& fFs;
bool fIsFix;
size_t COMPRESSED_CHUNK_SIZE;
private:
private:
};
}

1
writeengine/shared/we_define.h
View File

@ -348,6 +348,7 @@ const int ERR_COMP_READ_FILE = ERR_COMPBASE + 16;// Failed to read from a
const int ERR_COMP_WRITE_FILE = ERR_COMPBASE + 17;// Failed to write to a compresssed data file
const int ERR_COMP_CLOSE_FILE = ERR_COMPBASE + 18;// Failed to close a compressed data file
const int ERR_COMP_TRUNCATE_ZERO = ERR_COMPBASE + 19;// Invalid attempt to truncate file to 0 bytes
const int ERR_COMP_WRONG_COMP_TYPE = ERR_COMPBASE + 20;// Invalid compression type.
//--------------------------------------------------------------------------
// Auto-increment error

174
writeengine/shared/we_fileop.cpp
View File

@ -652,14 +652,19 @@ int FileOp::extendFile(
// @bug 5349: check that new extent's fbo is not past current EOF
if (m_compressionType)
{
char hdrsIn[ compress::IDBCompressInterface::HDR_BUF_LEN * 2 ];
char hdrsIn[ compress::CompressInterface::HDR_BUF_LEN * 2 ];
RETURN_ON_ERROR( readHeaders(pFile, hdrsIn) );
IDBCompressInterface compressor;
unsigned int ptrCount = compressor.getPtrCount(hdrsIn);
std::unique_ptr<compress::CompressInterface> compressor(
compress::getCompressInterfaceByType(
compress::CompressInterface::getCompressionType(hdrsIn)));
unsigned int ptrCount =
compress::CompressInterface::getPtrCount(hdrsIn);
unsigned int chunkIndex = 0;
unsigned int blockOffsetWithinChunk = 0;
compressor.locateBlock((hwm - 1), chunkIndex, blockOffsetWithinChunk);
compressor->locateBlock((hwm - 1), chunkIndex,
blockOffsetWithinChunk);
//std::ostringstream oss1;
//oss1 << "Extending compressed column file"<<
@ -816,9 +821,8 @@ int FileOp::extendFile(
if ((m_compressionType) && (hdrs))
{
IDBCompressInterface compressor;
compressor.initHdr(hdrs, width, colDataType, m_compressionType);
compressor.setLBIDByIndex(hdrs, startLbid, 0);
compress::CompressInterface::initHdr(hdrs, width, colDataType, m_compressionType);
compress::CompressInterface::setLBIDByIndex(hdrs, startLbid, 0);
}
}
@ -976,9 +980,8 @@ int FileOp::addExtentExactFile(
if ((m_compressionType) && (hdrs))
{
IDBCompressInterface compressor;
compressor.initHdr(hdrs, width, colDataType, m_compressionType);
compressor.setLBIDByIndex(hdrs, startLbid, 0);
compress::CompressInterface::initHdr(hdrs, width, colDataType, m_compressionType);
compress::CompressInterface::setLBIDByIndex(hdrs, startLbid, 0);
}
}
@ -1064,13 +1067,11 @@ int FileOp::initColumnExtent(
{
if ((bNewFile) && (m_compressionType))
{
char hdrs[IDBCompressInterface::HDR_BUF_LEN * 2];
IDBCompressInterface compressor;
compressor.initHdr(hdrs, width, colDataType, m_compressionType);
compressor.setLBIDByIndex(hdrs, lbid, 0);
char hdrs[CompressInterface::HDR_BUF_LEN * 2];
compress::CompressInterface::initHdr(hdrs, width, colDataType, m_compressionType);
compress::CompressInterface::setLBIDByIndex(hdrs, lbid, 0);
if (bAbbrevExtent)
compressor.setBlockCount(hdrs, nBlocks);
compress::CompressInterface::setBlockCount(hdrs, nBlocks);
RETURN_ON_ERROR(writeHeaders(pFile, hdrs));
}
@ -1262,7 +1263,7 @@ int FileOp::initAbbrevCompColumnExtent(
Stats::startParseEvent(WE_STATS_COMPRESS_COL_INIT_ABBREV_EXT);
#endif
char hdrs[IDBCompressInterface::HDR_BUF_LEN * 2];
char hdrs[CompressInterface::HDR_BUF_LEN * 2];
rc = writeInitialCompColumnChunk( pFile,
nBlocks,
INITIAL_EXTENT_ROWS_TO_DISK,
@ -1308,24 +1309,30 @@ int FileOp::writeInitialCompColumnChunk(
execplan::CalpontSystemCatalog::ColDataType colDataType,
char* hdrs)
{
const int INPUT_BUFFER_SIZE = nRows * width;
const size_t INPUT_BUFFER_SIZE = nRows * width;
char* toBeCompressedInput = new char[INPUT_BUFFER_SIZE];
unsigned int userPaddingBytes = Config::getNumCompressedPadBlks() *
BYTE_PER_BLOCK;
const int OUTPUT_BUFFER_SIZE = IDBCompressInterface::maxCompressedSize(INPUT_BUFFER_SIZE) +
userPaddingBytes;
// Compress an initialized abbreviated extent
// Initially m_compressionType == 0, but this function is used under
// condtion where m_compressionType > 0.
std::unique_ptr<CompressInterface> compressor(
compress::getCompressInterfaceByType(m_compressionType,
userPaddingBytes));
const size_t OUTPUT_BUFFER_SIZE =
compressor->maxCompressedSize(INPUT_BUFFER_SIZE) + userPaddingBytes +
compress::CompressInterface::COMPRESSED_CHUNK_INCREMENT_SIZE;
unsigned char* compressedOutput = new unsigned char[OUTPUT_BUFFER_SIZE];
unsigned int outputLen = OUTPUT_BUFFER_SIZE;
size_t outputLen = OUTPUT_BUFFER_SIZE;
boost::scoped_array<char> toBeCompressedInputPtr( toBeCompressedInput );
boost::scoped_array<unsigned char> compressedOutputPtr(compressedOutput);
setEmptyBuf( (unsigned char*)toBeCompressedInput,
INPUT_BUFFER_SIZE, emptyVal, width);
// Compress an initialized abbreviated extent
IDBCompressInterface compressor( userPaddingBytes );
int rc = compressor.compressBlock(toBeCompressedInput,
INPUT_BUFFER_SIZE, compressedOutput, outputLen );
int rc = compressor->compressBlock(toBeCompressedInput, INPUT_BUFFER_SIZE,
compressedOutput, outputLen);
if (rc != 0)
{
@ -1333,8 +1340,8 @@ int FileOp::writeInitialCompColumnChunk(
}
// Round up the compressed chunk size
rc = compressor.padCompressedChunks( compressedOutput,
outputLen, OUTPUT_BUFFER_SIZE );
rc = compressor->padCompressedChunks(compressedOutput, outputLen,
OUTPUT_BUFFER_SIZE);
if (rc != 0)
{
@ -1347,23 +1354,22 @@ int FileOp::writeInitialCompColumnChunk(
// "; blkAllocCnt: " << nBlocksAllocated <<
// "; compressedByteCnt: " << outputLen << std::endl;
compressor.initHdr(hdrs, width, colDataType, m_compressionType);
compressor.setBlockCount(hdrs, nBlocksAllocated);
compressor.setLBIDByIndex(hdrs, startLBID, 0);
compress::CompressInterface::initHdr(hdrs, width, colDataType, m_compressionType);
compress::CompressInterface::setBlockCount(hdrs, nBlocksAllocated);
compress::CompressInterface::setLBIDByIndex(hdrs, startLBID, 0);
// Store compression pointers in the header
std::vector<uint64_t> ptrs;
ptrs.push_back( IDBCompressInterface::HDR_BUF_LEN * 2 );
ptrs.push_back( outputLen + (IDBCompressInterface::HDR_BUF_LEN * 2) );
compressor.storePtrs(ptrs, hdrs);
ptrs.push_back( CompressInterface::HDR_BUF_LEN * 2 );
ptrs.push_back( outputLen + (CompressInterface::HDR_BUF_LEN * 2) );
compress::CompressInterface::storePtrs(ptrs, hdrs);
RETURN_ON_ERROR( writeHeaders(pFile, hdrs) );
// Write the compressed data
if ( pFile->write( compressedOutput, outputLen ) != outputLen )
{
size_t writtenLen = pFile->write(compressedOutput, outputLen);
if (writtenLen != outputLen)
return ERR_FILE_WRITE;
}
return NO_ERROR;
}
@ -1421,7 +1427,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
return ERR_FILE_OPEN;
}
char hdrs[ IDBCompressInterface::HDR_BUF_LEN * 2 ];
char hdrs[ CompressInterface::HDR_BUF_LEN * 2 ];
rc = readHeaders( pFile, hdrs );
if (rc != NO_ERROR)
@ -1432,9 +1438,14 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
}
int userPadBytes = Config::getNumCompressedPadBlks() * BYTE_PER_BLOCK;
IDBCompressInterface compressor( userPadBytes );
std::unique_ptr<CompressInterface> compressor(
compress::getCompressInterfaceByType(
compress::CompressInterface::getCompressionType(hdrs),
userPadBytes));
CompChunkPtrList chunkPtrs;
int rcComp = compressor.getPtrList( hdrs, chunkPtrs );
int rcComp = compress::CompressInterface::getPtrList(hdrs, chunkPtrs);
if (rcComp != 0)
{
@ -1444,7 +1455,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
}
// Nothing to do if the proposed HWM is < the current block count
uint64_t blkCount = compressor.getBlockCount(hdrs);
uint64_t blkCount = compress::CompressInterface::getBlockCount(hdrs);
if (blkCount > (hwm + 1))
{
@ -1455,7 +1466,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
const unsigned int ROWS_PER_EXTENT =
BRMWrapper::getInstance()->getInstance()->getExtentRows();
const unsigned int ROWS_PER_CHUNK =
IDBCompressInterface::UNCOMPRESSED_INBUF_LEN / colWidth;
CompressInterface::UNCOMPRESSED_INBUF_LEN / colWidth;
const unsigned int CHUNKS_PER_EXTENT = ROWS_PER_EXTENT / ROWS_PER_CHUNK;
// If this is an abbreviated extent, we first expand to a full extent
@ -1493,7 +1504,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
CompChunkPtr chunkOutPtr;
rc = expandAbbrevColumnChunk( pFile, emptyVal, colWidth,
chunkPtrs[0], chunkOutPtr );
chunkPtrs[0], chunkOutPtr, hdrs );
if (rc != NO_ERROR)
{
@ -1515,7 +1526,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
// Update block count to reflect a full extent
blkCount = (ROWS_PER_EXTENT * colWidth) / BYTE_PER_BLOCK;
compressor.setBlockCount( hdrs, blkCount );
compress::CompressInterface::setBlockCount(hdrs, blkCount);
}
// Calculate the number of empty chunks we need to add to fill this extent
@ -1532,7 +1543,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
compressor.getBlockCount(hdrs) << std::endl;
std::cout << "Pointer Header Size (in bytes): " <<
(compressor.getHdrSize(hdrs) -
IDBCompressInterface::HDR_BUF_LEN) << std::endl;
CompressInterface::HDR_BUF_LEN) << std::endl;
std::cout << "Chunk Pointers (offset,length): " << std::endl;
for (unsigned k = 0; k < chunkPtrs.size(); k++)
@ -1551,8 +1562,10 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
// Fill in or add necessary remaining empty chunks
if (numChunksToFill > 0)
{
const int IN_BUF_LEN = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
const int OUT_BUF_LEN = IDBCompressInterface::maxCompressedSize(IN_BUF_LEN) + userPadBytes;
const int IN_BUF_LEN = CompressInterface::UNCOMPRESSED_INBUF_LEN;
const int OUT_BUF_LEN =
compressor->maxCompressedSize(IN_BUF_LEN) + userPadBytes +
compress::CompressInterface::COMPRESSED_CHUNK_INCREMENT_SIZE;
// Allocate buffer, and store in scoped_array to insure it's deletion.
// Create scope {...} to manage deletion of buffers
@ -1566,9 +1579,9 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
// Compress and then pad the compressed chunk
setEmptyBuf( (unsigned char*)toBeCompressedBuf,
IN_BUF_LEN, emptyVal, colWidth );
unsigned int outputLen = OUT_BUF_LEN;
rcComp = compressor.compressBlock( toBeCompressedBuf,
IN_BUF_LEN, compressedBuf, outputLen );
size_t outputLen = OUT_BUF_LEN;
rcComp = compressor->compressBlock(toBeCompressedBuf, IN_BUF_LEN,
compressedBuf, outputLen);
if (rcComp != 0)
{
@ -1579,8 +1592,8 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
toBeCompressedInputPtr.reset(); // release memory
rcComp = compressor.padCompressedChunks( compressedBuf,
outputLen, OUT_BUF_LEN );
rcComp = compressor->padCompressedChunks(compressedBuf, outputLen,
OUT_BUF_LEN);
if (rcComp != 0)
{
@ -1639,7 +1652,7 @@ int FileOp::fillCompColumnExtentEmptyChunks(OID oid,
ptrs.push_back( chunkPtrs[chunkPtrs.size() - 1].first +
chunkPtrs[chunkPtrs.size() - 1].second );
compressor.storePtrs( ptrs, hdrs );
compress::CompressInterface::storePtrs(ptrs, hdrs);
rc = writeHeaders( pFile, hdrs );
@ -1697,11 +1710,24 @@ int FileOp::expandAbbrevColumnChunk(
const uint8_t* emptyVal,
int colWidth,
const CompChunkPtr& chunkInPtr,
CompChunkPtr& chunkOutPtr )
CompChunkPtr& chunkOutPtr,
const char *hdrs )
{
int userPadBytes = Config::getNumCompressedPadBlks() * BYTE_PER_BLOCK;
const int IN_BUF_LEN = IDBCompressInterface::UNCOMPRESSED_INBUF_LEN;
const int OUT_BUF_LEN = IDBCompressInterface::maxCompressedSize(IN_BUF_LEN) + userPadBytes;
auto realCompressionType = m_compressionType;
if (hdrs)
{
realCompressionType =
compress::CompressInterface::getCompressionType(hdrs);
}
std::unique_ptr<CompressInterface> compressor(
compress::getCompressInterfaceByType(realCompressionType,
userPadBytes));
const int IN_BUF_LEN = CompressInterface::UNCOMPRESSED_INBUF_LEN;
const int OUT_BUF_LEN =
compressor->maxCompressedSize(IN_BUF_LEN) + userPadBytes +
compress::CompressInterface::COMPRESSED_CHUNK_INCREMENT_SIZE;
char* toBeCompressedBuf = new char[ IN_BUF_LEN ];
boost::scoped_array<char> toBeCompressedPtr(toBeCompressedBuf);
@ -1717,13 +1743,10 @@ int FileOp::expandAbbrevColumnChunk(
chunkInPtr.second) );
// Uncompress an "abbreviated" chunk into our 4MB buffer
unsigned int outputLen = IN_BUF_LEN;
IDBCompressInterface compressor( userPadBytes );
int rc = compressor.uncompressBlock(
compressedInBuf,
chunkInPtr.second,
(unsigned char*)toBeCompressedBuf,
outputLen);
size_t outputLen = IN_BUF_LEN;
int rc = compressor->uncompressBlock(compressedInBuf, chunkInPtr.second,
(unsigned char*) toBeCompressedBuf,
outputLen);
if (rc != 0)
{
@ -1739,11 +1762,8 @@ int FileOp::expandAbbrevColumnChunk(
// Compress the data we just read, as a "full" 4MB chunk
outputLen = OUT_BUF_LEN;
rc = compressor.compressBlock(
reinterpret_cast<char*>(toBeCompressedBuf),
IN_BUF_LEN,
compressedOutBuf,
outputLen );
rc = compressor->compressBlock(reinterpret_cast<char*>(toBeCompressedBuf),
IN_BUF_LEN, compressedOutBuf, outputLen);
if (rc != 0)
{
@ -1751,8 +1771,8 @@ int FileOp::expandAbbrevColumnChunk(
}
// Round up the compressed chunk size
rc = compressor.padCompressedChunks( compressedOutBuf,
outputLen, OUT_BUF_LEN );
rc = compressor->padCompressedChunks(compressedOutBuf, outputLen,
OUT_BUF_LEN);
if (rc != 0)
{
@ -1782,7 +1802,7 @@ int FileOp::writeHeaders(IDBDataFile* pFile, const char* hdr) const
RETURN_ON_ERROR( setFileOffset(pFile, 0, SEEK_SET) );
// Write the headers
if (pFile->write( hdr, IDBCompressInterface::HDR_BUF_LEN * 2 ) != IDBCompressInterface::HDR_BUF_LEN * 2)
if (pFile->write( hdr, CompressInterface::HDR_BUF_LEN * 2 ) != CompressInterface::HDR_BUF_LEN * 2)
{
return ERR_FILE_WRITE;
}
@ -1808,7 +1828,7 @@ int FileOp::writeHeaders(IDBDataFile* pFile, const char* controlHdr,
RETURN_ON_ERROR( setFileOffset(pFile, 0, SEEK_SET) );
// Write the control header
if (pFile->write( controlHdr, IDBCompressInterface::HDR_BUF_LEN ) != IDBCompressInterface::HDR_BUF_LEN)
if (pFile->write( controlHdr, CompressInterface::HDR_BUF_LEN ) != CompressInterface::HDR_BUF_LEN)
{
return ERR_FILE_WRITE;
}
@ -2651,9 +2671,8 @@ int FileOp::readHeaders( IDBDataFile* pFile, char* hdrs ) const
{
RETURN_ON_ERROR( setFileOffset(pFile, 0) );
RETURN_ON_ERROR( readFile( pFile, reinterpret_cast<unsigned char*>(hdrs),
(IDBCompressInterface::HDR_BUF_LEN * 2) ) );
IDBCompressInterface compressor;
int rc = compressor.verifyHdr( hdrs );
(CompressInterface::HDR_BUF_LEN * 2) ) );
int rc = compress::CompressInterface::verifyHdr(hdrs);
if (rc != 0)
{
@ -2671,11 +2690,10 @@ int FileOp::readHeaders( IDBDataFile* pFile, char* hdr1, char* hdr2 ) const
unsigned char* hdrPtr = reinterpret_cast<unsigned char*>(hdr1);
RETURN_ON_ERROR( setFileOffset(pFile, 0) );
RETURN_ON_ERROR( readFile( pFile, hdrPtr,
IDBCompressInterface::HDR_BUF_LEN ));
CompressInterface::HDR_BUF_LEN ));
IDBCompressInterface compressor;
int ptrSecSize = compressor.getHdrSize(hdrPtr) -
IDBCompressInterface::HDR_BUF_LEN;
int ptrSecSize = compress::CompressInterface::getHdrSize(hdrPtr) -
CompressInterface::HDR_BUF_LEN;
return readFile( pFile, reinterpret_cast<unsigned char*>(hdr2),
ptrSecSize );
}

10
writeengine/shared/we_fileop.h
View File

@ -529,11 +529,11 @@ private:
FileOp(const FileOp& rhs);
FileOp& operator=(const FileOp& rhs);
int expandAbbrevColumnChunk( IDBDataFile* pFile,
const uint8_t* emptyVal,
int colWidth,
const compress::CompChunkPtr& chunkInPtr,
compress::CompChunkPtr& chunkOutPt);
int expandAbbrevColumnChunk(IDBDataFile* pFile, const uint8_t* emptyVal,
int colWidth,
const compress::CompChunkPtr& chunkInPtr,
compress::CompChunkPtr& chunkOutPt,
const char* hdrs = nullptr);
int initAbbrevCompColumnExtent(
IDBDataFile* pFile, uint16_t dbRoot, int nBlocks,

30
writeengine/shared/we_rbmetawriter.cpp
View File

@ -1007,9 +1007,9 @@ void RBMetaWriter::backupHWMChunk(
}
// Read Control header
char controlHdr[ IDBCompressInterface::HDR_BUF_LEN ];
char controlHdr[ CompressInterface::HDR_BUF_LEN ];
rc = fileOp.readFile( dbFile, (unsigned char*)controlHdr,
IDBCompressInterface::HDR_BUF_LEN );
CompressInterface::HDR_BUF_LEN );
if (rc != NO_ERROR)
{
@ -1025,8 +1025,7 @@ void RBMetaWriter::backupHWMChunk(
throw WeException( oss.str(), rc );
}
IDBCompressInterface compressor;
int rc1 = compressor.verifyHdr( controlHdr );
int rc1 = compress::CompressInterface::verifyHdr(controlHdr);
if (rc1 != 0)
{
@ -1045,9 +1044,23 @@ void RBMetaWriter::backupHWMChunk(
throw WeException( oss.str(), rc );
}
auto compressionType =
compress::CompressInterface::getCompressionType(controlHdr);
std::unique_ptr<compress::CompressInterface> compressor(
compress::getCompressInterfaceByType(compressionType));
if (!compressor)
{
WErrorCodes ec;
std::ostringstream oss;
oss << "Ivalid compression type " << compressionType;
fileOp.closeFile( dbFile );
throw WeException(oss.str(), rc);
}
// Read Pointer header data
uint64_t hdrSize = compressor.getHdrSize(controlHdr);
uint64_t ptrHdrSize = hdrSize - IDBCompressInterface::HDR_BUF_LEN;
uint64_t hdrSize = compress::CompressInterface::getHdrSize(controlHdr);
uint64_t ptrHdrSize = hdrSize - CompressInterface::HDR_BUF_LEN;
char* pointerHdr = new char[ptrHdrSize];
rc = fileOp.readFile( dbFile, (unsigned char*)pointerHdr, ptrHdrSize );
@ -1067,7 +1080,8 @@ void RBMetaWriter::backupHWMChunk(
}
CompChunkPtrList chunkPtrs;
rc = compressor.getPtrList(pointerHdr, ptrHdrSize, chunkPtrs );
rc = compress::CompressInterface::getPtrList(pointerHdr, ptrHdrSize,
chunkPtrs);
delete[] pointerHdr;
if (rc != 0)
@ -1087,7 +1101,7 @@ void RBMetaWriter::backupHWMChunk(
unsigned int blockOffsetWithinChunk = 0;
unsigned char* buffer = 0;
uint64_t chunkSize = 0;
compressor.locateBlock(startingHWM, chunkIndex, blockOffsetWithinChunk);
compressor->locateBlock(startingHWM, chunkIndex, blockOffsetWithinChunk);
if (chunkIndex < chunkPtrs.size())
{

10
writeengine/wrapper/we_colopcompress.cpp
View File

@ -121,9 +121,9 @@ int ColumnOpCompress0::saveBlock(IDBDataFile* pFile, const unsigned char* writeB
* Constructor
*/
ColumnOpCompress1::ColumnOpCompress1(Log* logger)
ColumnOpCompress1::ColumnOpCompress1(uint32_t compressionType, Log* logger)
{
m_compressionType = 1;
m_compressionType = compressionType;
m_chunkManager = new ChunkManager();
if (logger)
@ -164,11 +164,7 @@ bool ColumnOpCompress1::abbreviatedExtent(IDBDataFile* pFile, int colWidth) cons
int ColumnOpCompress1::blocksInFile(IDBDataFile* pFile) const
{
CompFileHeader compFileHeader;
readHeaders(pFile, compFileHeader.fControlData, compFileHeader.fPtrSection);
compress::IDBCompressInterface compressor;
return compressor.getBlockCount(compFileHeader.fControlData);
return m_chunkManager->getBlockCount(pFile);
}

2
writeengine/wrapper/we_colopcompress.h
View File

@ -97,7 +97,7 @@ public:
/**
* @brief Constructor
*/
EXPORT ColumnOpCompress1(Log* logger = 0);
EXPORT ColumnOpCompress1(uint32_t compressionType, Log* logger = 0);
/**
* @brief Default Destructor

4
writeengine/wrapper/we_dctnrycompress.cpp
View File

@ -67,9 +67,9 @@ DctnryCompress0::~DctnryCompress0()
/**
* Constructor
*/
DctnryCompress1::DctnryCompress1(Log* logger)
DctnryCompress1::DctnryCompress1(uint32_t compressionType, Log* logger)
{
m_compressionType = 1;
m_compressionType = compressionType;
m_chunkManager = new ChunkManager();
if (logger)

2
writeengine/wrapper/we_dctnrycompress.h
View File

@ -62,7 +62,7 @@ public:
/**
* @brief Constructor
*/
EXPORT DctnryCompress1(Log* logger = 0);
EXPORT DctnryCompress1(uint32_t compressionType, Log* logger = 0);
/**
* @brief Default Destructor

26
writeengine/wrapper/writeengine.cpp
View File

@ -76,19 +76,25 @@ StopWatch timer;
WriteEngineWrapper::WriteEngineWrapper() : m_opType(NOOP)
{
m_colOp[UN_COMPRESSED_OP] = new ColumnOpCompress0;
m_colOp[COMPRESSED_OP] = new ColumnOpCompress1;
m_dctnry[UN_COMPRESSED_OP] = new DctnryCompress0;
m_dctnry[COMPRESSED_OP] = new DctnryCompress1;
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_colOp[COMPRESSED_OP] = new ColumnOpCompress1;
m_dctnry[UN_COMPRESSED_OP] = new DctnryCompress0;
m_dctnry[COMPRESSED_OP] = new DctnryCompress1;
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
@ -96,9 +102,13 @@ WriteEngineWrapper::WriteEngineWrapper(const WriteEngineWrapper& rhs) : m_opTyp
WriteEngineWrapper::~WriteEngineWrapper()
{
delete m_colOp[UN_COMPRESSED_OP];
delete m_colOp[COMPRESSED_OP];
delete m_dctnry[UN_COMPRESSED_OP];
delete m_dctnry[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

53
writeengine/wrapper/writeengine.h
View File

@ -58,9 +58,10 @@ namespace WriteEngine
{
//... Total compression operation: un_compresssed, compressed
const int UN_COMPRESSED_OP = 0;
const int COMPRESSED_OP = 1;
const int TOTAL_COMPRESS_OP = 2;
const int UN_COMPRESSED_OP = 0;
const int COMPRESSED_OP_1 = 1;
const int COMPRESSED_OP_2 = 2;
const int TOTAL_COMPRESS_OP = 3;
//...Forward class declarations
class Log;
@ -446,8 +447,10 @@ public:
*/
void setIsInsert(bool bIsInsert)
{
m_colOp[COMPRESSED_OP]->chunkManager()->setIsInsert(bIsInsert);
m_dctnry[COMPRESSED_OP]->chunkManager()->setIsInsert(true);
m_colOp[COMPRESSED_OP_1]->chunkManager()->setIsInsert(bIsInsert);
m_dctnry[COMPRESSED_OP_1]->chunkManager()->setIsInsert(true);
m_colOp[COMPRESSED_OP_2]->chunkManager()->setIsInsert(bIsInsert);
m_dctnry[COMPRESSED_OP_2]->chunkManager()->setIsInsert(true);
}
/**
@ -458,7 +461,7 @@ public:
*/
bool getIsInsert()
{
return m_colOp[COMPRESSED_OP]->chunkManager()->getIsInsert();
return m_colOp[COMPRESSED_OP_1]->chunkManager()->getIsInsert();
}
std::tr1::unordered_map<TxnID, SP_TxnLBIDRec_t>& getTxnMap()
@ -475,10 +478,23 @@ public:
*/
int flushChunks(int rc, const std::map<FID, FID>& columOids)
{
int rtn1 = m_colOp[COMPRESSED_OP]->chunkManager()->flushChunks(rc, columOids);
int rtn2 = m_dctnry[COMPRESSED_OP]->chunkManager()->flushChunks(rc, columOids);
std::vector<int32_t> compressedOpIds = {COMPRESSED_OP_1,
COMPRESSED_OP_2};
return (rtn1 != NO_ERROR ? rtn1 : rtn2);
for (const auto compressedOpId : compressedOpIds)
{
auto rtn = m_colOp[compressedOpId]->chunkManager()->flushChunks(
rc, columOids);
if (rtn != NO_ERROR)
return rtn;
rtn = m_dctnry[compressedOpId]->chunkManager()->flushChunks(
rc, columOids);
if (rtn != NO_ERROR)
return rtn;
}
return NO_ERROR;
}
/**
@ -524,7 +540,7 @@ public:
int startTransaction(const TxnID& txnid)
{
int rc = 0;
rc = m_colOp[COMPRESSED_OP]->chunkManager()->startTransaction(txnid);
rc = m_colOp[COMPRESSED_OP_1]->chunkManager()->startTransaction(txnid);
//if ( rc == 0)
// rc = m_dctnry[COMPRESSED_OP]->chunkManager()->startTransaction(txnid);
return rc;
@ -537,7 +553,8 @@ public:
int confirmTransaction (const TxnID& txnid)
{
int rc = 0;
rc = m_colOp[COMPRESSED_OP]->chunkManager()->confirmTransaction (txnid);
rc = m_colOp[COMPRESSED_OP_1]->chunkManager()->confirmTransaction(
txnid);
return rc;
}
@ -549,7 +566,8 @@ public:
int endTransaction(const TxnID& txnid, bool success)
{
int rc = 0;
rc = m_colOp[COMPRESSED_OP]->chunkManager()->endTransaction(txnid, success);
rc = m_colOp[COMPRESSED_OP_1]->chunkManager()->endTransaction(txnid,
success);
//if ( rc == 0)
// rc = m_dctnry[COMPRESSED_OP]->chunkManager()->endTransaction(txnid, success);
return rc;
@ -785,7 +803,16 @@ private:
int op(int compressionType)
{
return (compressionType > 0 ? COMPRESSED_OP : UN_COMPRESSED_OP);
switch (compressionType)
{
case 1:
case 2:
return COMPRESSED_OP_1;
case 3:
return COMPRESSED_OP_2;
}
return 0;
}