xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

enh compress

This commit is contained in:
yihaoDeng 2024-06-22 11:35:47 +08:00
parent 9cab625f0b
commit 697ca8914e
5 changed files with 2354 additions and 423 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/common/src/tcol.c
View File

@ -250,7 +250,7 @@ bool checkColumnEncodeOrSetDefault(uint8_t type, char encode[TSDB_CL_COMPRESS_OP
strncpy(encode, getDefaultEncodeStr(type), TSDB_CL_COMPRESS_OPTION_LEN); strncpy(encode, getDefaultEncodeStr(type), TSDB_CL_COMPRESS_OPTION_LEN);
return true; return true;
} }
return checkColumnEncode(encode); return checkColumnEncode(encode) && validColEncode(type, columnEncodeVal(encode));
} }
bool checkColumnCompress(char compress[TSDB_CL_COMPRESS_OPTION_LEN]) { bool checkColumnCompress(char compress[TSDB_CL_COMPRESS_OPTION_LEN]) {
if (0 == strlen(compress)) return true; if (0 == strlen(compress)) return true;
@ -267,7 +267,8 @@ bool checkColumnCompressOrSetDefault(uint8_t type, char compress[TSDB_CL_COMPRES
strncpy(compress, getDefaultCompressStr(type), TSDB_CL_COMPRESS_OPTION_LEN); strncpy(compress, getDefaultCompressStr(type), TSDB_CL_COMPRESS_OPTION_LEN);
return true; return true;
} }
return checkColumnCompress(compress);
return checkColumnCompress(compress) && validColCompress(type, columnCompressVal(compress));
} }
bool checkColumnLevel(char level[TSDB_CL_COMPRESS_OPTION_LEN]) { bool checkColumnLevel(char level[TSDB_CL_COMPRESS_OPTION_LEN]) {
if (0 == strlen(level)) return true; if (0 == strlen(level)) return true;
@ -288,7 +289,7 @@ bool checkColumnLevelOrSetDefault(uint8_t type, char level[TSDB_CL_COMPRESS_OPTI
strncpy(level, getDefaultLevelStr(type), TSDB_CL_COMPRESS_OPTION_LEN); strncpy(level, getDefaultLevelStr(type), TSDB_CL_COMPRESS_OPTION_LEN);
return true; return true;
} }
return checkColumnLevel(level); return checkColumnLevel(level) && validColCompressLevel(type, columnLevelVal(level));
} }
void setColEncode(uint32_t* compress, uint8_t l1) { void setColEncode(uint32_t* compress, uint8_t l1) {

19
source/common/src/tglobal.c
View File

@ -240,8 +240,8 @@ int64_t tsTickPerMin[] = {60000L, 60000000L, 60000000000L};
int64_t tsTickPerHour[] = {3600000L, 3600000000L, 3600000000000L}; int64_t tsTickPerHour[] = {3600000L, 3600000000L, 3600000000000L};
// lossy compress 7 // lossy compress 7
char tsLossyColumns[32] = ""; // "float|double" means all float and double columns can be lossy compressed. set empty char tsLossyColumns[32] = "float|double"; // "float|double" means all float and double columns can be lossy compressed.
// can close lossy compress. // set empty can close lossy compress.
// below option can take effect when tsLossyColumns not empty // below option can take effect when tsLossyColumns not empty
float tsFPrecision = 1E-8; // float column precision float tsFPrecision = 1E-8; // float column precision
double tsDPrecision = 1E-16; // double column precision double tsDPrecision = 1E-16; // double column precision
@ -563,11 +563,12 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, CFG_SCOPE_BOTH, CFG_DYN_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, CFG_SCOPE_BOTH, CFG_DYN_BOTH) != 0) return -1;
if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 200000, CFG_SCOPE_BOTH, CFG_DYN_NONE) != 0) return -1; if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 200000, CFG_SCOPE_BOTH, CFG_DYN_NONE) != 0) return -1;
if (cfgAddBool(pCfg, "multiResultFunctionStarReturnTags", tsMultiResultFunctionStarReturnTags, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT) != 0) return -1; if (cfgAddBool(pCfg, "multiResultFunctionStarReturnTags", tsMultiResultFunctionStarReturnTags, CFG_SCOPE_CLIENT,
CFG_DYN_CLIENT) != 0)
return -1;
if (cfgAddInt32(pCfg, "countAlwaysReturnValue", tsCountAlwaysReturnValue, 0, 1, CFG_SCOPE_BOTH, CFG_DYN_CLIENT) != 0) if (cfgAddInt32(pCfg, "countAlwaysReturnValue", tsCountAlwaysReturnValue, 0, 1, CFG_SCOPE_BOTH, CFG_DYN_CLIENT) != 0)
return -1; return -1;
if (cfgAddInt32(pCfg, "maxTsmaCalcDelay", tsMaxTsmaCalcDelay, 600, 86400, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT) != if (cfgAddInt32(pCfg, "maxTsmaCalcDelay", tsMaxTsmaCalcDelay, 600, 86400, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT) != 0)
0)
return -1; return -1;
if (cfgAddInt32(pCfg, "tsmaDataDeleteMark", tsmaDataDeleteMark, 60 * 60 * 1000, INT64_MAX, CFG_SCOPE_CLIENT, if (cfgAddInt32(pCfg, "tsmaDataDeleteMark", tsmaDataDeleteMark, 60 * 60 * 1000, INT64_MAX, CFG_SCOPE_CLIENT,
CFG_DYN_CLIENT) != 0) CFG_DYN_CLIENT) != 0)
@ -618,7 +619,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
tsNumOfSupportVnodes = tsNumOfCores * 2; tsNumOfSupportVnodes = tsNumOfCores * 2;
tsNumOfSupportVnodes = TMAX(tsNumOfSupportVnodes, 2); tsNumOfSupportVnodes = TMAX(tsNumOfSupportVnodes, 2);
if (cfgAddInt32(pCfg, "supportVnodes", tsNumOfSupportVnodes, 0, 4096, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "supportVnodes", tsNumOfSupportVnodes, 0, 4096, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0)
return -1;
if (cfgAddString(pCfg, "encryptAlgorithm", tsEncryptAlgorithm, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddString(pCfg, "encryptAlgorithm", tsEncryptAlgorithm, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
if (cfgAddString(pCfg, "encryptScope", tsEncryptScope, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddString(pCfg, "encryptScope", tsEncryptScope, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
@ -749,8 +751,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "tmqMaxTopicNum", tmqMaxTopicNum, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) if (cfgAddInt32(pCfg, "tmqMaxTopicNum", tmqMaxTopicNum, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0)
return -1; return -1;
if (cfgAddInt32(pCfg, "tmqRowSize", tmqRowSize, 1, 1000000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) if (cfgAddInt32(pCfg, "tmqRowSize", tmqRowSize, 1, 1000000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1;
return -1;
if (cfgAddInt32(pCfg, "maxTsmaNum", tsMaxTsmaNum, 0, 3, CFG_SCOPE_SERVER, CFG_DYN_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "maxTsmaNum", tsMaxTsmaNum, 0, 3, CFG_SCOPE_SERVER, CFG_DYN_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
@ -807,7 +808,6 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
CFG_DYN_ENT_SERVER) != 0) CFG_DYN_ENT_SERVER) != 0)
return -1; return -1;
if (cfgAddString(pCfg, "lossyColumns", tsLossyColumns, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
if (cfgAddFloat(pCfg, "fPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddFloat(pCfg, "fPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
if (cfgAddFloat(pCfg, "dPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddFloat(pCfg, "dPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
if (cfgAddInt32(pCfg, "maxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddInt32(pCfg, "maxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
@ -1268,7 +1268,6 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32; tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32;
tstrncpy(tsLossyColumns, cfgGetItem(pCfg, "lossyColumns")->str, sizeof(tsLossyColumns));
tsFPrecision = cfgGetItem(pCfg, "fPrecision")->fval; tsFPrecision = cfgGetItem(pCfg, "fPrecision")->fval;
tsDPrecision = cfgGetItem(pCfg, "dPrecision")->fval; tsDPrecision = cfgGetItem(pCfg, "dPrecision")->fval;
tsMaxRange = cfgGetItem(pCfg, "maxRange")->i32; tsMaxRange = cfgGetItem(pCfg, "maxRange")->i32;

8
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
View File

@ -18,12 +18,10 @@
#include "dmNodes.h" #include "dmNodes.h"
#include "index.h" #include "index.h"
#include "qworker.h" #include "qworker.h"
#include "tstream.h"
#ifdef TD_TSZ
#include "tcompression.h" #include "tcompression.h"
#include "tglobal.h" #include "tglobal.h"
#include "tgrant.h" #include "tgrant.h"
#endif #include "tstream.h"
static bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper) { static bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper) {
SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper); SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);
@ -48,10 +46,8 @@ int32_t dmInitDnode(SDnode *pDnode) {
goto _OVER; goto _OVER;
} }
#ifdef TD_TSZ
// compress module init // compress module init
tsCompressInit(tsLossyColumns, tsFPrecision, tsDPrecision, tsMaxRange, tsCurRange, (int)tsIfAdtFse, tsCompressor); tsCompressInit(tsLossyColumns, tsFPrecision, tsDPrecision, tsMaxRange, tsCurRange, (int)tsIfAdtFse, tsCompressor);
#endif
pDnode->wrappers[DNODE].func = dmGetMgmtFunc(); pDnode->wrappers[DNODE].func = dmGetMgmtFunc();
pDnode->wrappers[MNODE].func = mmGetMgmtFunc(); pDnode->wrappers[MNODE].func = mmGetMgmtFunc();
@ -120,10 +116,8 @@ void dmCleanupDnode(SDnode *pDnode) {
indexCleanup(); indexCleanup();
taosConvDestroy(); taosConvDestroy();
#ifdef TD_TSZ
// compress destroy // compress destroy
tsCompressExit(); tsCompressExit();
#endif
dDebug("dnode is closed, ptr:%p", pDnode); dDebug("dnode is closed, ptr:%p", pDnode);
} }

23
source/util/src/tcompression.c
View File

@ -61,9 +61,7 @@
#include "zstd.h" #include "zstd.h"
#endif #endif
#ifdef TD_TSZ
#include "td_sz.h" #include "td_sz.h"
#endif
int32_t tsCompressPlain2(const char *const input, const int32_t nelements, char *const output, const char type); int32_t tsCompressPlain2(const char *const input, const int32_t nelements, char *const output, const char type);
int32_t tsDecompressPlain2(const char *const input, const int32_t nelements, char *const output, const char type); int32_t tsDecompressPlain2(const char *const input, const int32_t nelements, char *const output, const char type);
@ -322,7 +320,6 @@ static const int32_t TEST_NUMBER = 1;
#define safeInt64Add(a, b) (((a >= 0) && (b <= INT64_MAX - a)) || ((a < 0) && (b >= INT64_MIN - a))) #define safeInt64Add(a, b) (((a >= 0) && (b <= INT64_MAX - a)) || ((a < 0) && (b >= INT64_MIN - a)))
#ifdef TD_TSZ
bool lossyFloat = false; bool lossyFloat = false;
bool lossyDouble = false; bool lossyDouble = false;
@ -341,8 +338,6 @@ int32_t tsCompressInit(char *lossyColumns, float fPrecision, double dPrecision,
// exit call // exit call
void tsCompressExit() { tdszExit(); } void tsCompressExit() { tdszExit(); }
#endif
/* /*
* Compress Integer (Simple8B). * Compress Integer (Simple8B).
*/ */
@ -1214,7 +1209,6 @@ int32_t tsDecompressFloatImp(const char *const input, const int32_t nelements, c
return nelements * FLOAT_BYTES; return nelements * FLOAT_BYTES;
} }
#ifdef TD_TSZ
// //
// ---------- float double lossy ----------- // ---------- float double lossy -----------
// //
@ -1283,7 +1277,6 @@ int32_t tsDecompressDoubleLossyImp(const char *input, int32_t compressedSize, co
// decompressed with sz // decompressed with sz
return tdszDecompress(SZ_DOUBLE, input + 1, compressedSize - 1, nelements, output); return tdszDecompress(SZ_DOUBLE, input + 1, compressedSize - 1, nelements, output);
} }
#endif
#ifdef BUILD_NO_CALL #ifdef BUILD_NO_CALL
/************************************************************************* /*************************************************************************
@ -2463,13 +2456,11 @@ int32_t tsDecompressTimestamp(void *pIn, int32_t nIn, int32_t nEle, void *pOut,
// Float ===================================================== // Float =====================================================
int32_t tsCompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf, int32_t tsCompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf) { int32_t nBuf) {
#ifdef TD_TSZ
// lossy mode // lossy mode
if (lossyFloat) { if (lossyFloat) {
return tsCompressFloatLossyImp(pIn, nEle, pOut); return tsCompressFloatLossyImp(pIn, nEle, pOut);
// lossless mode // lossless mode
} else { } else {
#endif
if (cmprAlg == ONE_STAGE_COMP) { if (cmprAlg == ONE_STAGE_COMP) {
return tsCompressFloatImp(pIn, nEle, pOut); return tsCompressFloatImp(pIn, nEle, pOut);
} else if (cmprAlg == TWO_STAGE_COMP) { } else if (cmprAlg == TWO_STAGE_COMP) {
@ -2479,19 +2470,15 @@ int32_t tsCompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_
ASSERTS(0, "compress algo invalid"); ASSERTS(0, "compress algo invalid");
return -1; return -1;
} }
#ifdef TD_TSZ
} }
#endif
} }
int32_t tsDecompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf, int32_t tsDecompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf) { int32_t nBuf) {
#ifdef TD_TSZ
if (HEAD_ALGO(((uint8_t *)pIn)[0]) == ALGO_SZ_LOSSY) { if (HEAD_ALGO(((uint8_t *)pIn)[0]) == ALGO_SZ_LOSSY) {
// decompress lossy // decompress lossy
return tsDecompressFloatLossyImp(pIn, nIn, nEle, pOut); return tsDecompressFloatLossyImp(pIn, nIn, nEle, pOut);
} else { } else {
#endif
// decompress lossless // decompress lossless
if (cmprAlg == ONE_STAGE_COMP) { if (cmprAlg == ONE_STAGE_COMP) {
return tsDecompressFloatImp(pIn, nEle, pOut); return tsDecompressFloatImp(pIn, nEle, pOut);
@ -2502,20 +2489,16 @@ int32_t tsDecompressFloat(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int3
ASSERTS(0, "compress algo invalid"); ASSERTS(0, "compress algo invalid");
return -1; return -1;
} }
#ifdef TD_TSZ
} }
#endif
} }
// Double ===================================================== // Double =====================================================
int32_t tsCompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf, int32_t tsCompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf) { int32_t nBuf) {
#ifdef TD_TSZ
if (lossyDouble) { if (lossyDouble) {
// lossy mode // lossy mode
return tsCompressDoubleLossyImp(pIn, nEle, pOut); return tsCompressDoubleLossyImp(pIn, nEle, pOut);
} else { } else {
#endif
// lossless mode // lossless mode
if (cmprAlg == ONE_STAGE_COMP) { if (cmprAlg == ONE_STAGE_COMP) {
return tsCompressDoubleImp(pIn, nEle, pOut); return tsCompressDoubleImp(pIn, nEle, pOut);
@ -2526,19 +2509,15 @@ int32_t tsCompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32
ASSERTS(0, "compress algo invalid"); ASSERTS(0, "compress algo invalid");
return -1; return -1;
} }
#ifdef TD_TSZ
} }
#endif
} }
int32_t tsDecompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf, int32_t tsDecompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf) { int32_t nBuf) {
#ifdef TD_TSZ
if (HEAD_ALGO(((uint8_t *)pIn)[0]) == ALGO_SZ_LOSSY) { if (HEAD_ALGO(((uint8_t *)pIn)[0]) == ALGO_SZ_LOSSY) {
// decompress lossy // decompress lossy
return tsDecompressDoubleLossyImp(pIn, nIn, nEle, pOut); return tsDecompressDoubleLossyImp(pIn, nIn, nEle, pOut);
} else { } else {
#endif
// decompress lossless // decompress lossless
if (cmprAlg == ONE_STAGE_COMP) { if (cmprAlg == ONE_STAGE_COMP) {
return tsDecompressDoubleImp(pIn, nEle, pOut); return tsDecompressDoubleImp(pIn, nEle, pOut);
@ -2549,9 +2528,7 @@ int32_t tsDecompressDouble(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int
ASSERTS(0, "compress algo invalid"); ASSERTS(0, "compress algo invalid");
return -1; return -1;
} }
#ifdef TD_TSZ
} }
#endif
} }
// Binary ===================================================== // Binary =====================================================

2186
utils/test/c/tsz_test.c
View File

File diff suppressed because it is too large Load Diff