refactor code

This commit is contained in:
yihaoDeng 2024-03-27 05:49:13 +00:00
parent 2c5a5c4f28
commit 6d19283d1f
2 changed files with 96 additions and 46 deletions

View File

@ -256,7 +256,7 @@ typedef struct {
} TCompressPara;
typedef enum L1Compress {
L1_UNKNOWN,
L1_UNKNOWN = 0,
L1_SIMPLE_8B,
L1_XOR,
L1_RLE,
@ -264,7 +264,7 @@ typedef enum L1Compress {
} EL1CompressFuncType;
typedef enum L2Compress {
L2_UNKNOWN,
L2_UNKNOWN = 0,
L2_LZ4,
L2_ZLIB,
L2_ZSTD,

View File

@ -2667,33 +2667,39 @@ int32_t tsDecompressBigint(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int
}
}
#define FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmrlAlg, pBuf, nBuf, type, compress) \
#define FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, alg, pBuf, nBuf, type, compress) \
do { \
DEFINE_VAR(cmprAlg) \
DEFINE_VAR(alg) \
if (l1 != L1_DISABLED && l2 == L2_DISABLED) { \
if (compress) { \
uTrace("encode:%s, compress:%s, level:%s", compressL1Dict[l1].name, "disabled", "disabled"); \
uTrace("encode:%s, compress:%s, level:%s, type:%s", compressL1Dict[l1].name, "disabled", "disabled", \
tDataTypes[type].name); \
return compressL1Dict[l1].comprFn(pIn, nEle, pOut, type); \
} else { \
uTrace("dencode:%s, compress:%s, level:%s", compressL1Dict[l1].name, "disabled", "disabled"); \
uTrace("dencode:%s, compress:%s, level:%s, type:%s", compressL1Dict[l1].name, "disabled", "disabled", \
tDataTypes[type].name); \
return compressL1Dict[l1].decomprFn(pIn, nEle, pOut, type); \
} \
} else if (l1 != L1_DISABLED && l2 != L2_DISABLED) { \
if (compress) { \
uTrace("encode:%s, compress:%s, level:%d", compressL1Dict[l1].name, compressL2Dict[l2].name, lvl); \
uTrace("encode:%s, compress:%s, level:%d, type:%s, l1:%d", compressL1Dict[l1].name, compressL2Dict[l2].name, \
lvl, tDataTypes[type].name, l1); \
int32_t len = compressL1Dict[l1].comprFn(pIn, nEle, pBuf, type); \
return compressL2Dict[l2].comprFn(pBuf, len, pOut, nOut, type, lvl); \
} else { \
uTrace("dencode:%s, decompress:%s, level:%d", compressL1Dict[l1].name, compressL2Dict[l2].name, lvl); \
uTrace("dencode:%s, decompress:%s, level:%d, type:%s", compressL1Dict[l1].name, compressL2Dict[l2].name, lvl, \
tDataTypes[type].name); \
if (compressL2Dict[l2].decomprFn(pIn, nIn, pBuf, nBuf, type) < 0) return -1; \
return compressL1Dict[l1].decomprFn(pBuf, nEle, pOut, type); \
} \
} else if (l1 == L1_DISABLED && l2 != L2_DISABLED) { \
if (compress) { \
uTrace("encode:%s, compress:%s, level:%d", "disabled", compressL2Dict[l1].name, lvl); \
uTrace("encode:%s, compress:%s, level:%d, type:%s", "disabled", compressL2Dict[l1].name, lvl, \
tDataTypes[type].name); \
return compressL2Dict[l2].comprFn(pIn, nIn, pOut, nOut, type, lvl); \
} else { \
uTrace("dencode:%s, dcompress:%s, level:%d", "disabled", compressL2Dict[l1].name, lvl); \
uTrace("dencode:%s, dcompress:%s, level:%d, type:%s", "disabled", compressL2Dict[l1].name, lvl, \
tDataTypes[type].name); \
return compressL2Dict[l2].decomprFn(pIn, nIn, pOut, nOut, type); \
} \
} else { \
@ -2779,45 +2785,89 @@ int32_t tsDecompressString2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, in
// Bool =====================================================
int32_t tsCompressBool2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg, void *pBuf,
int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_BOOL, 1);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != 4) {
SET_COMPRESS(4, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_BOOL, 1);
}
int32_t tsDecompressBool2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg, void *pBuf,
int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_BOOL, 0);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != 4) {
SET_COMPRESS(4, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_BOOL, 0);
}
// Tinyint =====================================================
int32_t tsCompressTinyint2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg, void *pBuf,
int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_TINYINT, 1);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_TINYINT, 1);
}
int32_t tsDecompressTinyint2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg,
void *pBuf, int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_TINYINT, 0);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_TINYINT, 0);
}
// Smallint =====================================================
int32_t tsCompressSmallint2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg,
void *pBuf, int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_SMALLINT, 1);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_SMALLINT, 1);
}
int32_t tsDecompressSmallint2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg,
void *pBuf, int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_SMALLINT, 0);
uint32_t tCmprAlg = 0;
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_SMALLINT, 0);
}
// Int =====================================================
int32_t tsCompressInt2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg, void *pBuf,
int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_INT, 1);
uint32_t tCmprAlg = 0;
{
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_INT, 1);
}
int32_t tsDecompressInt2(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint32_t cmprAlg, void *pBuf,
int32_t nBuf) {
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, cmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_INT, 0);
uint32_t tCmprAlg = 0;
{
DEFINE_VAR(cmprAlg)
if (l1 != L1_XOR) {
SET_COMPRESS(L1_XOR, l2, lvl, tCmprAlg);
}
}
FUNC_COMPRESS_IMPL(pIn, nIn, nEle, pOut, nOut, tCmprAlg, pBuf, nBuf, TSDB_DATA_TYPE_INT, 0);
}
// Bigint =====================================================
@ -2841,9 +2891,9 @@ int32_t tsCompressImpl(int8_t type, void *pIn, int32_t nIn, int32_t nEle, void *
if (tsFindCompressAlg(type, cmprAlg, &fn1, &fn2)) return -1;
int32_t len = 0;
int8_t l1 = COMPRESS_L1_TYPE_U8(cmprAlg);
int8_t l2 = COMPRESS_L2_TYPE_U8(cmprAlg);
int8_t lvl = COMPRESS_L2_TYPE_LEVEL_U8(cmprAlg);
uint8_t l1 = COMPRESS_L1_TYPE_U8(cmprAlg);
uint8_t l2 = COMPRESS_L2_TYPE_U8(cmprAlg);
uint8_t lvl = COMPRESS_L2_TYPE_LEVEL_U8(cmprAlg);
if (l2 == L2_DISABLED) {
len = fn1.comprFn(pIn, nEle, pOut, type);