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

more code

This commit is contained in:
Hongze Cheng 2024-02-28 16:49:44 +08:00
parent 81d6a1479d
commit a8f95f53b7
7 changed files with 478 additions and 446 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
include/common/tdataformat.h
View File

@ -100,10 +100,10 @@ const static uint8_t BIT2_MAP[4] = {0b11111100, 0b11110011, 0b11001111, 0b001111
typedef struct { typedef struct {
int8_t cmprAlg; // filled by caller int8_t cmprAlg; // filled by caller
int8_t type; int8_t type;
int32_t originalDataSize; int32_t dataOriginalSize;
int32_t compressedDataSize; int32_t dataCompressedSize;
int32_t originalOffsetSize; int32_t offsetOriginalSize;
int32_t compressedOffsetSize; int32_t offsetCompressedSize;
} SValueColumnCompressInfo; } SValueColumnCompressInfo;
int32_t tValueColumnInit(SValueColumn *valCol); int32_t tValueColumnInit(SValueColumn *valCol);
@ -176,10 +176,9 @@ uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg); int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg);
extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull); extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull);
int32_t tColDataCompress(SColData *colData, SColDataCompressInfo *info, void *output, int32_t outputSize, int32_t tColDataCompress(SColData *colData, SColDataCompressInfo *info, SBuffer *output, SBuffer *assist);
SBuffer *buffer); int32_t tColDataDecompress(void *input, int32_t inputSize, SColDataCompressInfo *info, SColData *colData,
int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompressInfo *info, SColData *colData, SBuffer *assist);
SBuffer *helperBuffer);
// for stmt bind // for stmt bind
int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind, int32_t buffMaxLen); int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind, int32_t buffMaxLen);
@ -356,8 +355,7 @@ int32_t tDecompressData(void *input, // input
SBuffer *buffer // assistant buffer provided by caller, can be NULL SBuffer *buffer // assistant buffer provided by caller, can be NULL
); );
int32_t tCompressDataToBuffer(void *input, int32_t inputSize, SCompressInfo *info, SBuffer *output, SBuffer *assist); int32_t tCompressDataToBuffer(void *input, int32_t inputSize, SCompressInfo *info, SBuffer *output, SBuffer *assist);
int32_t tDecompressDataToBuffer(void *input, int32_t inputSize, const SCompressInfo *info, SBuffer *output, int32_t tDecompressDataToBuffer(void *input, int32_t inputSize, SCompressInfo *info, SBuffer *output, SBuffer *assist);
SBuffer *assist);
#endif #endif

353
include/util/tencode.h
View File

@ -468,217 +468,272 @@ static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) {
} }
// =========================================== // ===========================================
#define TPUT(TYPE, BUF, VAL, FORWARD) \ #define tPutV(p, v) \
do { \
if (BUF) { \
if (FORWARD) { \
*(TYPE*)(BUF) = (VAL); \
} else { \
*(TYPE*)((BUF) - sizeof(TYPE)) = (VAL); \
} \
} \
return sizeof(TYPE); \
} while (0)
#define TGET(TYPE, BUF, VAL, FORWARD) \
do { \
if (FORWARD) { \
*(TYPE*)(VAL) = *(TYPE*)(BUF); \
} else { \
*(TYPE*)(VAL) = *(TYPE*)((BUF) - sizeof(TYPE)); \
} \
return sizeof(TYPE); \
} while (0)
#define TPUTV(BUF, VAL, FORWARD) \
do { \ do { \
int32_t n = 0; \ int32_t n = 0; \
for (;;) { \ for (;;) { \
if ((VAL) < 0x80) { \ if (v <= 0x7f) { \
if (BUF) { \ if (p) p[n] = v; \
if (FORWARD) { \
(BUF)[n] = (VAL); \
} else { \
(BUF)[-(n + 1)] = (VAL); \
} \
} \
n++; \ n++; \
break; \ break; \
} else { \
if (BUF) { \
if (FORWARD) { \
(BUF)[n] = ((VAL) & 0x7f) | 0x80; \
} else { \
(BUF)[-(n + 1)] = ((VAL) & 0x7f) | 0x80; \
} \
} \ } \
if (p) p[n] = (v & 0x7f) | 0x80; \
n++; \ n++; \
(VAL) >>= 7; \ v >>= 7; \
} \
} \
return n; \
} while (0)
#define TGETV(BUF, VAL, FORWARD) \
do { \
int32_t n = 0; \
if (VAL) { \
*(VAL) = 0; \
} \
for (;;) { \
if (FORWARD) { \
if (VAL) { \
(*(VAL)) |= ((BUF)[n] & 0x7f) << (7 * n); \
} \
if ((BUF)[n++] < 0x80) break; \
} else { \
if (VAL) { \
(*(VAL)) |= ((BUF)[-(n + 1)] & 0x7f) << (7 * n); \
} \
if ((BUF)[-(n++)] < 0x80) break; \
} \
} \ } \
return n; \ return n; \
} while (0) } while (0)
// PUT // PUT
static FORCE_INLINE int32_t tPutU8(uint8_t* p, uint8_t v, bool forward) { TPUT(uint8_t, p, v, forward); } static FORCE_INLINE int32_t tPutU8(uint8_t* p, uint8_t v) {
static FORCE_INLINE int32_t tPutI8(uint8_t* p, int8_t v, bool forward) { TPUT(int8_t, p, v, forward); } if (p) ((uint8_t*)p)[0] = v;
static FORCE_INLINE int32_t tPutU16(uint8_t* p, uint16_t v, bool forward) { TPUT(uint16_t, p, v, forward); } return sizeof(uint8_t);
static FORCE_INLINE int32_t tPutI16(uint8_t* p, int16_t v, bool forward) { TPUT(int16_t, p, v, forward); } }
static FORCE_INLINE int32_t tPutU32(uint8_t* p, uint32_t v, bool forward) { TPUT(uint32_t, p, v, forward); }
static FORCE_INLINE int32_t tPutI32(uint8_t* p, int32_t v, bool forward) { TPUT(int32_t, p, v, forward); }
static FORCE_INLINE int32_t tPutU64(uint8_t* p, uint64_t v, bool forward) { TPUT(uint64_t, p, v, forward); }
static FORCE_INLINE int32_t tPutI64(uint8_t* p, int64_t v, bool forward) { TPUT(int64_t, p, v, forward); }
static FORCE_INLINE int32_t tPutFloat(uint8_t* p, float f, bool forward) { static FORCE_INLINE int32_t tPutI8(uint8_t* p, int8_t v) {
if (p) ((int8_t*)p)[0] = v;
return sizeof(int8_t);
}
static FORCE_INLINE int32_t tPutU16(uint8_t* p, uint16_t v) {
if (p) ((uint16_t*)p)[0] = v;
return sizeof(uint16_t);
}
static FORCE_INLINE int32_t tPutI16(uint8_t* p, int16_t v) {
if (p) ((int16_t*)p)[0] = v;
return sizeof(int16_t);
}
static FORCE_INLINE int32_t tPutU32(uint8_t* p, uint32_t v) {
if (p) ((uint32_t*)p)[0] = v;
return sizeof(uint32_t);
}
static FORCE_INLINE int32_t tPutI32(uint8_t* p, int32_t v) {
if (p) ((int32_t*)p)[0] = v;
return sizeof(int32_t);
}
static FORCE_INLINE int32_t tPutU64(uint8_t* p, uint64_t v) {
if (p) ((uint64_t*)p)[0] = v;
return sizeof(uint64_t);
}
static FORCE_INLINE int32_t tPutI64(uint8_t* p, int64_t v) {
if (p) ((int64_t*)p)[0] = v;
return sizeof(int64_t);
}
static FORCE_INLINE int32_t tPutFloat(uint8_t* p, float f) {
union { union {
uint32_t ui; uint32_t ui;
float f; float f;
} v = {.f = f}; } v;
return tPutU32(p, v.ui, forward); v.f = f;
return tPutU32(p, v.ui);
} }
static FORCE_INLINE int32_t tPutDouble(uint8_t* p, double d, bool forward) { static FORCE_INLINE int32_t tPutDouble(uint8_t* p, double d) {
union { union {
uint64_t ui; uint64_t ui;
double d; double d;
} v = {.d = d}; } v;
return tPutU64(p, v.ui, forward); v.d = d;
return tPutU64(p, v.ui);
} }
static FORCE_INLINE int32_t tPutU16v(uint8_t* p, uint16_t v, bool forward) { TPUTV(p, v, forward); } static FORCE_INLINE int32_t tPutU16v(uint8_t* p, uint16_t v) { tPutV(p, v); }
static FORCE_INLINE int32_t tPutI16v(uint8_t* p, int16_t v, bool forward) {
return tPutU16v(p, ZIGZAGE(int16_t, v), forward); static FORCE_INLINE int32_t tPutI16v(uint8_t* p, int16_t v) { return tPutU16v(p, ZIGZAGE(int16_t, v)); }
}
static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v, bool forward) { TPUTV(p, v, forward); } static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v) { tPutV(p, v); }
static FORCE_INLINE int32_t tPutI32v(uint8_t* p, int32_t v, bool forward) {
return tPutU32v(p, ZIGZAGE(int32_t, v), forward); static FORCE_INLINE int32_t tPutI32v(uint8_t* p, int32_t v) { return tPutU32v(p, ZIGZAGE(int32_t, v)); }
}
static FORCE_INLINE int32_t tPutU64v(uint8_t* p, uint64_t v, bool forward) { TPUTV(p, v, forward); } static FORCE_INLINE int32_t tPutU64v(uint8_t* p, uint64_t v) { tPutV(p, v); }
static FORCE_INLINE int32_t tPutI64v(uint8_t* p, int64_t v, bool forward) {
return tPutU64v(p, ZIGZAGE(int64_t, v), forward); static FORCE_INLINE int32_t tPutI64v(uint8_t* p, int64_t v) { return tPutU64v(p, ZIGZAGE(int64_t, v)); }
}
// GET // GET
static FORCE_INLINE int32_t tGetU8(uint8_t* p, uint8_t* v, bool forward) { TGET(uint8_t, p, v, forward); } static FORCE_INLINE int32_t tGetU8(uint8_t* p, uint8_t* v) {
static FORCE_INLINE int32_t tGetI8(uint8_t* p, int8_t* v, bool forward) { TGET(int8_t, p, v, forward); } if (v) *v = ((uint8_t*)p)[0];
static FORCE_INLINE int32_t tGetU16(uint8_t* p, uint16_t* v, bool forward) { TGET(uint16_t, p, v, forward); } return sizeof(uint8_t);
static FORCE_INLINE int32_t tGetI16(uint8_t* p, int16_t* v, bool forward) { TGET(int16_t, p, v, forward); } }
static FORCE_INLINE int32_t tGetU32(uint8_t* p, uint32_t* v, bool forward) { TGET(uint32_t, p, v, forward); }
static FORCE_INLINE int32_t tGetI32(uint8_t* p, int32_t* v, bool forward) { TGET(int32_t, p, v, forward); } static FORCE_INLINE int32_t tGetI8(uint8_t* p, int8_t* v) {
static FORCE_INLINE int32_t tGetU64(uint8_t* p, uint64_t* v, bool forward) { TGET(uint64_t, p, v, forward); } if (v) *v = ((int8_t*)p)[0];
static FORCE_INLINE int32_t tGetI64(uint8_t* p, int64_t* v, bool forward) { TGET(int64_t, p, v, forward); } return sizeof(int8_t);
static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v, bool forward) { TGETV(p, v, forward); } }
static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v, bool forward) {
static FORCE_INLINE int32_t tGetU16(uint8_t* p, uint16_t* v) {
if (v) *v = ((uint16_t*)p)[0];
return sizeof(uint16_t);
}
static FORCE_INLINE int32_t tGetI16(uint8_t* p, int16_t* v) {
if (v) *v = ((int16_t*)p)[0];
return sizeof(int16_t);
}
static FORCE_INLINE int32_t tGetU32(uint8_t* p, uint32_t* v) {
if (v) *v = ((uint32_t*)p)[0];
return sizeof(uint32_t);
}
static FORCE_INLINE int32_t tGetI32(uint8_t* p, int32_t* v) {
if (v) *v = ((int32_t*)p)[0];
return sizeof(int32_t);
}
static FORCE_INLINE int32_t tGetU64(uint8_t* p, uint64_t* v) {
if (v) *v = ((uint64_t*)p)[0];
return sizeof(uint64_t);
}
static FORCE_INLINE int32_t tGetI64(uint8_t* p, int64_t* v) {
if (v) *v = ((int64_t*)p)[0];
return sizeof(int64_t);
}
static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v) {
int32_t n = 0;
if (v) *v = 0;
for (;;) {
if (p[n] <= 0x7f) {
if (v) (*v) |= (((uint16_t)p[n]) << (7 * n));
n++;
break;
}
if (v) (*v) |= (((uint16_t)(p[n] & 0x7f)) << (7 * n));
n++;
}
return n;
}
static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
int32_t n;
uint16_t tv; uint16_t tv;
int32_t n = tGetU16v(p, &tv, forward);
if (v) { n = tGetU16v(p, &tv);
*v = ZIGZAGD(int16_t, tv); if (v) *v = ZIGZAGD(int16_t, tv);
}
return n; return n;
} }
static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v, bool forward) { TGETV(p, v, forward); }
static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v, bool forward) { static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) {
int32_t n = 0;
if (v) *v = 0;
for (;;) {
if (p[n] <= 0x7f) {
if (v) (*v) |= (((uint32_t)p[n]) << (7 * n));
n++;
break;
}
if (v) (*v) |= (((uint32_t)(p[n] & 0x7f)) << (7 * n));
n++;
}
return n;
}
static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
int32_t n;
uint32_t tv; uint32_t tv;
int32_t n = tGetU32v(p, &tv, forward);
if (v) { n = tGetU32v(p, &tv);
*v = ZIGZAGD(int32_t, tv); if (v) *v = ZIGZAGD(int32_t, tv);
}
return n; return n;
} }
static FORCE_INLINE int32_t tGetU64v(uint8_t* p, uint64_t* v, bool forward) { TGETV(p, v, forward); }
static FORCE_INLINE int32_t tGetI64v(uint8_t* p, int64_t* v, bool forward) { static FORCE_INLINE int32_t tGetU64v(uint8_t* p, uint64_t* v) {
int32_t n = 0;
if (v) *v = 0;
for (;;) {
if (p[n] <= 0x7f) {
if (v) (*v) |= (((uint64_t)p[n]) << (7 * n));
n++;
break;
}
if (v) (*v) |= (((uint64_t)(p[n] & 0x7f)) << (7 * n));
n++;
}
return n;
}
static FORCE_INLINE int32_t tGetI64v(uint8_t* p, int64_t* v) {
int32_t n;
uint64_t tv; uint64_t tv;
int32_t n = tGetU64v(p, &tv, forward);
if (v) { n = tGetU64v(p, &tv);
*v = ZIGZAGD(int64_t, tv); if (v) *v = ZIGZAGD(int64_t, tv);
}
return n; return n;
} }
static FORCE_INLINE int32_t tGetFloat(uint8_t* p, float* f, bool forward) {
static FORCE_INLINE int32_t tGetFloat(uint8_t* p, float* f) {
int32_t n = 0;
union { union {
uint32_t ui; uint32_t ui;
float f; float f;
} v; } v;
int32_t n = tGetU32(p, &v.ui, forward); n = tGetU32(p, &v.ui);
if (f) {
*f = v.f; *f = v.f;
}
return n; return n;
} }
static FORCE_INLINE int32_t tGetDouble(uint8_t* p, double* d, bool forward) { static FORCE_INLINE int32_t tGetDouble(uint8_t* p, double* d) {
int32_t n = 0;
union { union {
uint64_t ui; uint64_t ui;
double d; double d;
} v; } v;
int32_t n = tGetU64(p, &v.ui, forward); n = tGetU64(p, &v.ui);
if (d) {
*d = v.d; *d = v.d;
}
return n; return n;
} }
// ===================== // =====================
static FORCE_INLINE int32_t tPutBinary(uint8_t* p, uint8_t* pData, uint32_t nData, bool forward) { static FORCE_INLINE int32_t tPutBinary(uint8_t* p, uint8_t* pData, uint32_t nData) {
int32_t n = tPutU32v(p, nData, forward); int n = 0;
if (p) {
if (forward) { n += tPutU32v(p ? p + n : p, nData);
memcpy(p + n, pData, nData); if (p) memcpy(p + n, pData, nData);
} else {
memcpy(p - n - nData, pData, nData);
}
}
n += nData; n += nData;
return n; return n;
} }
static FORCE_INLINE int32_t tGetBinary(uint8_t* p, uint8_t** ppData, uint32_t* nData, bool forward) { static FORCE_INLINE int32_t tGetBinary(uint8_t* p, uint8_t** ppData, uint32_t* nData) {
int32_t n = 0;
uint32_t nt; uint32_t nt;
int32_t n = tGetU32v(p, &nt, forward);
if (nData) { n += tGetU32v(p, &nt);
*nData = nt; if (nData) *nData = nt;
} if (ppData) *ppData = p + n;
if (ppData) { n += nt;
if (forward) {
*ppData = p + n; return n;
} else {
*ppData = p - n - nt;
}
}
return n + nt;
} }
static FORCE_INLINE int32_t tPutCStr(uint8_t* p, char* pData, bool forward) { static FORCE_INLINE int32_t tPutCStr(uint8_t* p, char* pData) {
return tPutBinary(p, (uint8_t*)pData, strlen(pData) + 1, forward); return tPutBinary(p, (uint8_t*)pData, strlen(pData) + 1);
}
static FORCE_INLINE int32_t tGetCStr(uint8_t* p, char** ppData, bool forward) {
return tGetBinary(p, (uint8_t**)ppData, NULL, forward);
} }
static FORCE_INLINE int32_t tGetCStr(uint8_t* p, char** ppData) { return tGetBinary(p, (uint8_t**)ppData, NULL); }
#ifdef __cplusplus #ifdef __cplusplus
} }

255
source/common/src/tdataformat.c
View File

@ -93,21 +93,20 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
flag |= HAS_VALUE; flag |= HAS_VALUE;
maxIdx = nkv; maxIdx = nkv;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ntp = ntp + tPutU32v(NULL, pColVal->value.nData, true) + pColVal->value.nData; ntp = ntp + tPutU32v(NULL, pColVal->value.nData) + pColVal->value.nData;
nkv = nkv + tPutI16v(NULL, pTColumn->colId, true) + tPutU32v(NULL, pColVal->value.nData, true) + nkv = nkv + tPutI16v(NULL, pTColumn->colId) + tPutU32v(NULL, pColVal->value.nData) + pColVal->value.nData;
pColVal->value.nData;
} else { } else {
nkv = nkv + tPutI16v(NULL, pTColumn->colId, true) + pTColumn->bytes; nkv = nkv + tPutI16v(NULL, pTColumn->colId) + pTColumn->bytes;
} }
if (pTColumn->flags & COL_IS_KEY) { if (pTColumn->flags & COL_IS_KEY) {
flag |= HAS_MULTI_KEY; flag |= HAS_MULTI_KEY;
numPrimaryKeyCols++; numPrimaryKeyCols++;
ntp += (tPutI8(NULL, pTColumn->type, false) // type ntp += (tPutI8(NULL, pTColumn->type) // type
+ tPutI32v(NULL, pTColumn->offset, false) // offset + tPutI32v(NULL, pTColumn->offset) // offset
); );
nkv += (tPutI8(NULL, pTColumn->type, false) // type nkv += (tPutI8(NULL, pTColumn->type) // type
+ tPutI32v(NULL, nIdx, false) // index + tPutI32v(NULL, nIdx) // index
); );
} }
@ -119,7 +118,7 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
ASSERT((pTColumn->flags & COL_IS_KEY) == 0); ASSERT((pTColumn->flags & COL_IS_KEY) == 0);
flag |= HAS_NULL; flag |= HAS_NULL;
maxIdx = nkv; maxIdx = nkv;
nkv += tPutI16v(NULL, -pTColumn->colId, true); nkv += tPutI16v(NULL, -pTColumn->colId);
nIdx++; nIdx++;
} else { } else {
if (ASSERTS(0, "invalid input")) { if (ASSERTS(0, "invalid input")) {
@ -171,9 +170,9 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
} }
} }
if (flag & HAS_MULTI_KEY) { if (flag & HAS_MULTI_KEY) {
ntp += (tPutI8(NULL, numPrimaryKeyCols, false) // numPrimaryKeyCols ntp += (tPutI8(NULL, numPrimaryKeyCols) // numPrimaryKeyCols
+ tPutU32v(NULL, pTSchema->numOfCols, false) // numOfCols + tPutU32v(NULL, pTSchema->numOfCols) // numOfCols
+ tPutU32v(NULL, pTSchema->flen, false) // fixedLen + tPutU32v(NULL, pTSchema->flen) // fixedLen
); );
} }
// Key-Value Row Format // Key-Value Row Format
@ -188,7 +187,7 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
flag |= KV_FLG_BIG; flag |= KV_FLG_BIG;
} }
if (flag & HAS_MULTI_KEY) { if (flag & HAS_MULTI_KEY) {
nkv += tPutI8(NULL, numPrimaryKeyCols, false); nkv += tPutI8(NULL, numPrimaryKeyCols);
} }
int32_t rowLength; int32_t rowLength;
if (nkv < ntp) { if (nkv < ntp) {
@ -225,9 +224,9 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
uint8_t *pEnd = ((uint8_t *)pRow) + rowLength; uint8_t *pEnd = ((uint8_t *)pRow) + rowLength;
if (pRow->flag & HAS_MULTI_KEY) { if (pRow->flag & HAS_MULTI_KEY) {
pEnd -= tPutI8(pEnd, numPrimaryKeyCols, false); // numPrimaryKeyCols pEnd -= tPutI8(pEnd, numPrimaryKeyCols); // numPrimaryKeyCols
pEnd -= tPutU32v(pEnd, pTSchema->numOfCols, false); // numOfCols pEnd -= tPutU32v(pEnd, pTSchema->numOfCols); // numOfCols
pEnd -= tPutU32v(NULL, pTSchema->flen, false); // fixedLen pEnd -= tPutU32v(NULL, pTSchema->flen); // fixedLen
} }
if (flag >> 4) { // KV if (flag >> 4) { // KV
SKVIdx *pIdx = (SKVIdx *)pRow->data; SKVIdx *pIdx = (SKVIdx *)pRow->data;
@ -255,9 +254,9 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv; ((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv;
} }
nv += tPutI16v(pv + nv, pTColumn->colId, true); nv += tPutI16v(pv + nv, pTColumn->colId);
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
nv += tPutU32v(pv + nv, pColVal->value.nData, true); nv += tPutU32v(pv + nv, pColVal->value.nData);
memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData); memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData);
nv += pColVal->value.nData; nv += pColVal->value.nData;
} else { } else {
@ -266,8 +265,8 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
} }
if (pRow->flag & HAS_MULTI_KEY) { if (pRow->flag & HAS_MULTI_KEY) {
pEnd -= tPutI8(pEnd, pTColumn->type, false); pEnd -= tPutI8(pEnd, pTColumn->type);
pEnd -= tPutI32v(pEnd, iIdx, false); pEnd -= tPutI32v(pEnd, iIdx);
} }
iIdx++; iIdx++;
@ -279,7 +278,7 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
} else { } else {
((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv; ((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv;
} }
nv += tPutI16v(pv + nv, -pTColumn->colId, true); nv += tPutI16v(pv + nv, -pTColumn->colId);
iIdx++; iIdx++;
} }
@ -343,7 +342,7 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
*(int32_t *)(pf + pTColumn->offset) = nv; *(int32_t *)(pf + pTColumn->offset) = nv;
nv += tPutU32v(pv + nv, pColVal->value.nData, true); nv += tPutU32v(pv + nv, pColVal->value.nData);
if (pColVal->value.nData) { if (pColVal->value.nData) {
memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData); memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData);
nv += pColVal->value.nData; nv += pColVal->value.nData;
@ -353,8 +352,8 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
} }
if (pTColumn->flags & COL_IS_KEY) { if (pTColumn->flags & COL_IS_KEY) {
pEnd -= tPutI8(pEnd, pTColumn->type, false); pEnd -= tPutI8(pEnd, pTColumn->type);
pEnd -= tPutI32v(pEnd, pTColumn->offset, false); pEnd -= tPutI32v(pEnd, pTColumn->offset);
} }
} else if (COL_VAL_IS_NONE(pColVal)) { // NONE } else if (COL_VAL_IS_NONE(pColVal)) { // NONE
ROW_SET_BITMAP(pb, flag, iTColumn - 1, BIT_FLG_NONE); ROW_SET_BITMAP(pb, flag, iTColumn - 1, BIT_FLG_NONE);
@ -440,7 +439,7 @@ int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal)
} }
int16_t cid; int16_t cid;
pData += tGetI16v(pData, &cid, true); pData += tGetI16v(pData, &cid);
if (TABS(cid) == pTColumn->colId) { if (TABS(cid) == pTColumn->colId) {
if (cid < 0) { if (cid < 0) {
@ -451,7 +450,7 @@ int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal)
pColVal->flag = CV_FLAG_VALUE; pColVal->flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pData += tGetU32v(pData, &pColVal->value.nData, true); pData += tGetU32v(pData, &pColVal->value.nData);
if (pColVal->value.nData > 0) { if (pColVal->value.nData > 0) {
pColVal->value.pData = pData; pColVal->value.pData = pData;
} else { } else {
@ -477,7 +476,7 @@ int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal)
pColVal->flag = CV_FLAG_VALUE; pColVal->flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
uint8_t *pData = pRow->data + pTSchema->flen + *(int32_t *)(pRow->data + pTColumn->offset); uint8_t *pData = pRow->data + pTSchema->flen + *(int32_t *)(pRow->data + pTColumn->offset);
pData += tGetU32v(pData, &pColVal->value.nData, true); pData += tGetU32v(pData, &pColVal->value.nData);
if (pColVal->value.nData) { if (pColVal->value.nData) {
pColVal->value.pData = pData; pColVal->value.pData = pData;
} else { } else {
@ -530,7 +529,7 @@ int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal)
pColVal->flag = CV_FLAG_VALUE; pColVal->flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
uint8_t *pData = pv + *(int32_t *)(pf + pTColumn->offset); uint8_t *pData = pv + *(int32_t *)(pf + pTColumn->offset);
pData += tGetU32v(pData, &pColVal->value.nData, true); pData += tGetU32v(pData, &pColVal->value.nData);
if (pColVal->value.nData) { if (pColVal->value.nData) {
pColVal->value.pData = pData; pColVal->value.pData = pData;
} else { } else {
@ -793,7 +792,7 @@ SColVal *tRowIterNext(SRowIter *pIter) {
} }
int16_t cid; int16_t cid;
pData += tGetI16v(pData, &cid, true); pData += tGetI16v(pData, &cid);
if (TABS(cid) == pTColumn->colId) { if (TABS(cid) == pTColumn->colId) {
if (cid < 0) { if (cid < 0) {
@ -804,7 +803,7 @@ SColVal *tRowIterNext(SRowIter *pIter) {
pIter->cv.flag = CV_FLAG_VALUE; pIter->cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pData += tGetU32v(pData, &pIter->cv.value.nData, true); pData += tGetU32v(pData, &pIter->cv.value.nData);
if (pIter->cv.value.nData > 0) { if (pIter->cv.value.nData > 0) {
pIter->cv.value.pData = pData; pIter->cv.value.pData = pData;
} else { } else {
@ -863,7 +862,7 @@ SColVal *tRowIterNext(SRowIter *pIter) {
pIter->cv.flag = CV_FLAG_VALUE; pIter->cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
uint8_t *pData = pIter->pv + *(int32_t *)(pIter->pf + pTColumn->offset); uint8_t *pData = pIter->pv + *(int32_t *)(pIter->pf + pTColumn->offset);
pData += tGetU32v(pData, &pIter->cv.value.nData, true); pData += tGetU32v(pData, &pIter->cv.value.nData);
if (pIter->cv.value.nData > 0) { if (pIter->cv.value.nData > 0) {
pIter->cv.value.pData = pData; pIter->cv.value.pData = pData;
} else { } else {
@ -1006,7 +1005,7 @@ static int32_t tRowTupleUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *
if (IS_VAR_DATA_TYPE(pColData->type)) { if (IS_VAR_DATA_TYPE(pColData->type)) {
uint8_t *pData = pv + *(int32_t *)(pf + pTColumn->offset); uint8_t *pData = pv + *(int32_t *)(pf + pTColumn->offset);
uint32_t nData; uint32_t nData;
pData += tGetU32v(pData, &nData, true); pData += tGetU32v(pData, &nData);
if (flag == 0) { if (flag == 0) {
code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](pColData, pData, nData); code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](pColData, pData, nData);
} else { } else {
@ -1080,7 +1079,7 @@ static int32_t tRowKVUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aCo
} }
int16_t cid; int16_t cid;
pData += tGetI16v(pData, &cid, true); pData += tGetI16v(pData, &cid);
if (TABS(cid) == pTColumn->colId) { if (TABS(cid) == pTColumn->colId) {
if (cid < 0) { if (cid < 0) {
@ -1093,7 +1092,7 @@ static int32_t tRowKVUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aCo
} else { } else {
uint32_t nData; uint32_t nData;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pData += tGetU32v(pData, &nData, true); pData += tGetU32v(pData, &nData);
} else { } else {
nData = 0; nData = 0;
} }
@ -1160,7 +1159,7 @@ void tRowGetKey(SRow *pRow, SRowKey *key) {
} else { } else {
uint8_t *pEnd = ((uint8_t *)pRow) + pRow->len; uint8_t *pEnd = ((uint8_t *)pRow) + pRow->len;
pEnd -= tGetU8(pEnd, &key->numOfPKs, false); pEnd -= tGetU8(pEnd, &key->numOfPKs);
if (pRow->flag >> 4) { // Key-Value format if (pRow->flag >> 4) { // Key-Value format
SKVIdx *pKVIdx = (SKVIdx *)pRow->data; SKVIdx *pKVIdx = (SKVIdx *)pRow->data;
@ -1179,8 +1178,8 @@ void tRowGetKey(SRow *pRow, SRowKey *key) {
uint8_t *pData; uint8_t *pData;
SValue *pValue = &key->pks[iKey]; SValue *pValue = &key->pks[iKey];
pEnd -= tGetI8(pEnd, &pValue->type, false); pEnd -= tGetI8(pEnd, &pValue->type);
pEnd -= tGetI32v(pEnd, &index, false); pEnd -= tGetI32v(pEnd, &index);
if (pRow->flag & KV_FLG_LIT) { if (pRow->flag & KV_FLG_LIT) {
pData = pv + ((uint8_t *)pKVIdx->idx)[index]; pData = pv + ((uint8_t *)pKVIdx->idx)[index];
@ -1190,9 +1189,9 @@ void tRowGetKey(SRow *pRow, SRowKey *key) {
pData = pv + ((uint32_t *)pKVIdx->idx)[index]; pData = pv + ((uint32_t *)pKVIdx->idx)[index];
} }
pData += tGetI16v(pData, NULL, true); pData += tGetI16v(pData, NULL);
if (IS_VAR_DATA_TYPE(pValue->type)) { if (IS_VAR_DATA_TYPE(pValue->type)) {
pData += tGetU32v(pData, &pValue->nData, true); pData += tGetU32v(pData, &pValue->nData);
pValue->pData = pData; pValue->pData = pData;
} else { } else {
memcpy(&pValue->val, pData, TYPE_BYTES[pValue->type]); memcpy(&pValue->val, pData, TYPE_BYTES[pValue->type]);
@ -1204,8 +1203,8 @@ void tRowGetKey(SRow *pRow, SRowKey *key) {
uint32_t fixedLen; uint32_t fixedLen;
uint32_t numOfCols; uint32_t numOfCols;
pEnd -= tGetU32v(pEnd, &numOfCols, false); pEnd -= tGetU32v(pEnd, &numOfCols);
pEnd -= tGetU32v(pEnd, &fixedLen, false); pEnd -= tGetU32v(pEnd, &fixedLen);
switch (pRow->flag & (HAS_VALUE | HAS_NULL | HAS_NONE)) { switch (pRow->flag & (HAS_VALUE | HAS_NULL | HAS_NONE)) {
case (HAS_VALUE | HAS_NONE): case (HAS_VALUE | HAS_NONE):
@ -1225,12 +1224,12 @@ void tRowGetKey(SRow *pRow, SRowKey *key) {
int32_t offset; int32_t offset;
SValue *pValue = &key->pks[iKey]; SValue *pValue = &key->pks[iKey];
pEnd -= tGetI8(pEnd, &pValue->type, false); pEnd -= tGetI8(pEnd, &pValue->type);
pEnd -= tGetI32v(pEnd, &offset, false); pEnd -= tGetI32v(pEnd, &offset);
if (IS_VAR_DATA_TYPE(key->pks[iKey].type)) { if (IS_VAR_DATA_TYPE(key->pks[iKey].type)) {
pValue->pData = pv + *(int32_t *)(pf + offset); pValue->pData = pv + *(int32_t *)(pf + offset);
pValue->pData += tGetU32v(pValue->pData, &pValue->nData, true); pValue->pData += tGetU32v(pValue->pData, &pValue->nData);
} else { } else {
memcpy(&key->pks[iKey].val, pf + offset, TYPE_BYTES[key->pks[iKey].type]); memcpy(&key->pks[iKey].val, pf + offset, TYPE_BYTES[key->pks[iKey].type]);
} }
@ -1430,18 +1429,18 @@ static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
// key // key
if (isJson) { if (isJson) {
n += tPutCStr(p ? p + n : p, pTagVal->pKey, true); n += tPutCStr(p ? p + n : p, pTagVal->pKey);
} else { } else {
n += tPutI16v(p ? p + n : p, pTagVal->cid, true); n += tPutI16v(p ? p + n : p, pTagVal->cid);
ASSERTS(pTagVal->cid > 0, "Invalid tag cid:%" PRIi16, pTagVal->cid); ASSERTS(pTagVal->cid > 0, "Invalid tag cid:%" PRIi16, pTagVal->cid);
} }
// type // type
n += tPutI8(p ? p + n : p, pTagVal->type, true); n += tPutI8(p ? p + n : p, pTagVal->type);
// value // value
if (IS_VAR_DATA_TYPE(pTagVal->type)) { if (IS_VAR_DATA_TYPE(pTagVal->type)) {
n += tPutBinary(p ? p + n : p, pTagVal->pData, pTagVal->nData, true); n += tPutBinary(p ? p + n : p, pTagVal->pData, pTagVal->nData);
} else { } else {
p = p ? p + n : p; p = p ? p + n : p;
n += tDataTypes[pTagVal->type].bytes; n += tDataTypes[pTagVal->type].bytes;
@ -1455,17 +1454,17 @@ static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
// key // key
if (isJson) { if (isJson) {
n += tGetCStr(p + n, &pTagVal->pKey, true); n += tGetCStr(p + n, &pTagVal->pKey);
} else { } else {
n += tGetI16v(p + n, &pTagVal->cid, true); n += tGetI16v(p + n, &pTagVal->cid);
} }
// type // type
n += tGetI8(p + n, &pTagVal->type, true); n += tGetI8(p + n, &pTagVal->type);
// value // value
if (IS_VAR_DATA_TYPE(pTagVal->type)) { if (IS_VAR_DATA_TYPE(pTagVal->type)) {
n += tGetBinary(p + n, &pTagVal->pData, &pTagVal->nData, true); n += tGetBinary(p + n, &pTagVal->pData, &pTagVal->nData);
} else { } else {
memcpy(&(pTagVal->i64), p + n, tDataTypes[pTagVal->type].bytes); memcpy(&(pTagVal->i64), p + n, tDataTypes[pTagVal->type].bytes);
n += tDataTypes[pTagVal->type].bytes; n += tDataTypes[pTagVal->type].bytes;
@ -1695,7 +1694,7 @@ void tTagSetCid(const STag *pTag, int16_t iTag, int16_t cid) {
offset = pTag->idx[iTag]; offset = pTag->idx[iTag];
} }
tPutI16v(p + offset, cid, true); tPutI16v(p + offset, cid);
} }
// STSchema ======================================== // STSchema ========================================
@ -2635,20 +2634,12 @@ _exit:
return code; return code;
} }
int32_t tColDataCompress(SColData *colData, // column data int32_t tColDataCompress(SColData *colData, SColDataCompressInfo *info, SBuffer *output, SBuffer *assist) {
SColDataCompressInfo *info, // compress info
void *output, // output
int32_t outputSize, // output size
SBuffer *buffer // assistant buffer
) {
int32_t code; int32_t code;
SBuffer local; SBuffer local;
char *outputStart = output;
ASSERT(colData->nVal > 0); ASSERT(colData->nVal > 0);
tBufferInit(&local);
(*info) = (SColDataCompressInfo){ (*info) = (SColDataCompressInfo){
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
.columnFlag = colData->cflag, .columnFlag = colData->cflag,
@ -2659,12 +2650,12 @@ int32_t tColDataCompress(SColData *colData, // column data
}; };
if (colData->flag == HAS_NONE || colData->flag == HAS_NULL) { if (colData->flag == HAS_NONE || colData->flag == HAS_NULL) {
tBufferDestroy(&local);
return 0; return 0;
} }
if (buffer == NULL) { tBufferInit(&local);
buffer = &local; if (assist == NULL) {
assist = &local;
} }
// bitmap // bitmap
@ -2680,15 +2671,13 @@ int32_t tColDataCompress(SColData *colData, // column data
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
}; };
code = tCompressData(colData->pBitMap, info->bitmapOriginalSize, &cinfo, outputStart, outputSize, buffer); code = tCompressDataToBuffer(colData->pBitMap, info->bitmapOriginalSize, &cinfo, output, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
} }
info->bitmapCompressedSize = cinfo.compressedSize; info->bitmapCompressedSize = cinfo.compressedSize;
outputStart += cinfo.compressedSize;
outputSize -= cinfo.compressedSize;
} }
if (colData->flag == (HAS_NONE | HAS_NULL)) { if (colData->flag == (HAS_NONE | HAS_NULL)) {
@ -2705,15 +2694,13 @@ int32_t tColDataCompress(SColData *colData, // column data
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
}; };
code = tCompressData(colData->aOffset, info->offsetOriginalSize, &cinfo, outputStart, outputSize, buffer); code = tCompressDataToBuffer(colData->aOffset, info->offsetOriginalSize, &cinfo, output, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
} }
info->offsetCompressedSize = cinfo.compressedSize; info->offsetCompressedSize = cinfo.compressedSize;
outputStart += cinfo.compressedSize;
outputSize -= cinfo.compressedSize;
} }
// data // data
@ -2725,23 +2712,21 @@ int32_t tColDataCompress(SColData *colData, // column data
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
}; };
code = tCompressData(colData->pData, info->dataOriginalSize, &cinfo, outputStart, outputSize, buffer); code = tCompressDataToBuffer(colData->pData, info->dataOriginalSize, &cinfo, output, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
} }
info->dataCompressedSize = cinfo.compressedSize; info->dataCompressedSize = cinfo.compressedSize;
outputStart += cinfo.compressedSize;
outputSize -= cinfo.compressedSize;
} }
tBufferDestroy(&local); tBufferDestroy(&local);
return 0; return 0;
} }
int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompressInfo *info, SColData *colData, int32_t tColDataDecompress(void *input, int32_t inputSize, SColDataCompressInfo *info, SColData *colData,
SBuffer *buffer) { SBuffer *assist) {
int32_t code; int32_t code;
SBuffer local; SBuffer local;
char *inputStart = input; char *inputStart = input;
@ -2749,8 +2734,8 @@ int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompres
ASSERT(inputSize == info->bitmapCompressedSize + info->offsetCompressedSize + info->dataCompressedSize); ASSERT(inputSize == info->bitmapCompressedSize + info->offsetCompressedSize + info->dataCompressedSize);
tBufferInit(&local); tBufferInit(&local);
if (buffer == NULL) { if (assist == NULL) {
buffer = &local; assist = &local;
} }
tColDataClear(colData); tColDataClear(colData);
@ -2779,7 +2764,7 @@ int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompres
return code; return code;
} }
code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->pBitMap, cinfo.originalSize, buffer); code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->pBitMap, cinfo.originalSize, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
@ -2807,7 +2792,7 @@ int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompres
return code; return code;
} }
code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->aOffset, cinfo.originalSize, buffer); code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->aOffset, cinfo.originalSize, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
@ -2833,7 +2818,7 @@ int32_t tColDataDecompress(void *input, int32_t inputSize, const SColDataCompres
return code; return code;
} }
code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->pData, cinfo.originalSize, buffer); code = tDecompressData(inputStart, cinfo.compressedSize, &cinfo, colData->pData, cinfo.originalSize, assist);
if (code) { if (code) {
tBufferDestroy(&local); tBufferDestroy(&local);
return code; return code;
@ -3517,10 +3502,10 @@ _exit:
int32_t tPutColData(uint8_t *pBuf, SColData *pColData) { int32_t tPutColData(uint8_t *pBuf, SColData *pColData) {
int32_t n = 0; int32_t n = 0;
n += tPutI16v(pBuf ? pBuf + n : NULL, pColData->cid, true); n += tPutI16v(pBuf ? pBuf + n : NULL, pColData->cid);
n += tPutI8(pBuf ? pBuf + n : NULL, pColData->type, true); n += tPutI8(pBuf ? pBuf + n : NULL, pColData->type);
n += tPutI32v(pBuf ? pBuf + n : NULL, pColData->nVal, true); n += tPutI32v(pBuf ? pBuf + n : NULL, pColData->nVal);
n += tPutI8(pBuf ? pBuf + n : NULL, pColData->flag, true); n += tPutI8(pBuf ? pBuf + n : NULL, pColData->flag);
// bitmap // bitmap
switch (pColData->flag) { switch (pColData->flag) {
@ -3544,7 +3529,7 @@ int32_t tPutColData(uint8_t *pBuf, SColData *pColData) {
if (pBuf) memcpy(pBuf + n, pColData->aOffset, pColData->nVal << 2); if (pBuf) memcpy(pBuf + n, pColData->aOffset, pColData->nVal << 2);
n += (pColData->nVal << 2); n += (pColData->nVal << 2);
n += tPutI32v(pBuf ? pBuf + n : NULL, pColData->nData, true); n += tPutI32v(pBuf ? pBuf + n : NULL, pColData->nData);
if (pBuf) memcpy(pBuf + n, pColData->pData, pColData->nData); if (pBuf) memcpy(pBuf + n, pColData->pData, pColData->nData);
n += pColData->nData; n += pColData->nData;
} else { } else {
@ -3559,10 +3544,10 @@ int32_t tPutColData(uint8_t *pBuf, SColData *pColData) {
int32_t tGetColData(uint8_t *pBuf, SColData *pColData) { int32_t tGetColData(uint8_t *pBuf, SColData *pColData) {
int32_t n = 0; int32_t n = 0;
n += tGetI16v(pBuf + n, &pColData->cid, true); n += tGetI16v(pBuf + n, &pColData->cid);
n += tGetI8(pBuf + n, &pColData->type, true); n += tGetI8(pBuf + n, &pColData->type);
n += tGetI32v(pBuf + n, &pColData->nVal, true); n += tGetI32v(pBuf + n, &pColData->nVal);
n += tGetI8(pBuf + n, &pColData->flag, true); n += tGetI8(pBuf + n, &pColData->flag);
// bitmap // bitmap
switch (pColData->flag) { switch (pColData->flag) {
@ -3586,7 +3571,7 @@ int32_t tGetColData(uint8_t *pBuf, SColData *pColData) {
pColData->aOffset = (int32_t *)(pBuf + n); pColData->aOffset = (int32_t *)(pBuf + n);
n += (pColData->nVal << 2); n += (pColData->nVal << 2);
n += tGetI32v(pBuf + n, &pColData->nData, true); n += tGetI32v(pBuf + n, &pColData->nData);
pColData->pData = pBuf + n; pColData->pData = pBuf + n;
n += pColData->nData; n += pColData->nData;
} else { } else {
@ -4078,36 +4063,30 @@ int32_t tValueColumnGet(SValueColumn *valCol, int32_t idx, SValue *value) {
if (IS_VAR_DATA_TYPE(value->type)) { if (IS_VAR_DATA_TYPE(value->type)) {
int32_t offset, nextOffset; int32_t offset, nextOffset;
memcpy(&offset, (char *)tBufferGetData(&valCol->offsets) + idx * sizeof(offset), sizeof(offset)); tBufferGet(&valCol->offsets, idx, sizeof(offset), &offset);
if (idx == valCol->numOfValues - 1) { if (idx == valCol->numOfValues - 1) {
nextOffset = tBufferGetSize(&valCol->data); nextOffset = tBufferGetSize(&valCol->data);
} else { } else {
memcpy(&nextOffset, (char *)tBufferGetData(&valCol->offsets) + (idx + 1) * sizeof(nextOffset), tBufferGet(&valCol->offsets, idx + 1, sizeof(nextOffset), &nextOffset);
sizeof(nextOffset));
} }
value->nData = nextOffset - offset; value->nData = nextOffset - offset;
value->pData = (uint8_t *)((char *)tBufferGetData(&valCol->data) + offset); value->pData = (uint8_t *)tBufferGetDataAt(&valCol->data, offset);
} else { } else {
memcpy(&value->val, (char *)tBufferGetData(&valCol->data) + idx * tDataTypes[value->type].bytes, tBufferGet(&valCol->data, idx, tDataTypes[value->type].bytes, &value->val);
tDataTypes[value->type].bytes);
} }
return 0; return 0;
} }
int32_t tValueColumnCompress(SValueColumn *valCol, SValueColumnCompressInfo *info, SBuffer *output, SBuffer *assist) { int32_t tValueColumnCompress(SValueColumn *valCol, SValueColumnCompressInfo *info, SBuffer *output, SBuffer *assist) {
int32_t code; int32_t code;
SBuffer local;
ASSERT(valCol->numOfValues > 0);
(*info) = (SValueColumnCompressInfo){ (*info) = (SValueColumnCompressInfo){
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
.type = valCol->type, .type = valCol->type,
}; };
tBufferInit(&local);
if (assist == NULL) {
assist = &local;
}
// offset // offset
if (IS_VAR_DATA_TYPE(valCol->type)) { if (IS_VAR_DATA_TYPE(valCol->type)) {
SCompressInfo cinfo = { SCompressInfo cinfo = {
@ -4116,13 +4095,10 @@ int32_t tValueColumnCompress(SValueColumn *valCol, SValueColumnCompressInfo *inf
}; };
code = tCompressDataToBuffer(valCol->offsets.data, valCol->offsets.size, &cinfo, output, assist); code = tCompressDataToBuffer(valCol->offsets.data, valCol->offsets.size, &cinfo, output, assist);
if (code) { if (code) return code;
tBufferDestroy(&local);
return code;
}
info->originalOffsetSize = cinfo.originalSize; info->offsetOriginalSize = cinfo.originalSize;
info->compressedOffsetSize = cinfo.compressedSize; info->offsetCompressedSize = cinfo.compressedSize;
} }
// data // data
@ -4132,14 +4108,11 @@ int32_t tValueColumnCompress(SValueColumn *valCol, SValueColumnCompressInfo *inf
}; };
code = tCompressDataToBuffer(valCol->data.data, valCol->data.size, &cinfo, output, assist); code = tCompressDataToBuffer(valCol->data.data, valCol->data.size, &cinfo, output, assist);
if (code) { if (code) return code;
tBufferGetData(&local);
return code; info->dataOriginalSize = cinfo.originalSize;
} info->dataCompressedSize = cinfo.compressedSize;
info->originalDataSize = cinfo.originalSize;
info->compressedDataSize = cinfo.compressedSize;
tBufferDestroy(&local);
return 0; return 0;
} }
@ -4149,7 +4122,7 @@ int32_t tValueColumnDecompress(void *input, int32_t inputSize, const SValueColum
SBuffer local; SBuffer local;
char *inputStart = input; char *inputStart = input;
ASSERT(inputSize == info->compressedOffsetSize + info->compressedDataSize); ASSERT(inputSize == info->offsetCompressedSize + info->dataCompressedSize);
tValueColumnClear(valCol); tValueColumnClear(valCol);
tBufferInit(&local); tBufferInit(&local);
@ -4161,13 +4134,13 @@ int32_t tValueColumnDecompress(void *input, int32_t inputSize, const SValueColum
valCol->type = info->type; valCol->type = info->type;
// offset // offset
if (IS_VAR_DATA_TYPE(valCol->type)) { if (IS_VAR_DATA_TYPE(valCol->type)) {
valCol->numOfValues = info->originalOffsetSize / tDataTypes[TSDB_DATA_TYPE_INT].bytes; valCol->numOfValues = info->offsetOriginalSize / tDataTypes[TSDB_DATA_TYPE_INT].bytes;
SCompressInfo cinfo = { SCompressInfo cinfo = {
.dataType = TSDB_DATA_TYPE_INT, .dataType = TSDB_DATA_TYPE_INT,
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
.originalSize = info->originalOffsetSize, .originalSize = info->offsetOriginalSize,
.compressedSize = info->compressedOffsetSize, .compressedSize = info->offsetCompressedSize,
}; };
code = tBufferEnsureCapacity(&valCol->offsets, cinfo.originalSize); code = tBufferEnsureCapacity(&valCol->offsets, cinfo.originalSize);
@ -4184,15 +4157,15 @@ int32_t tValueColumnDecompress(void *input, int32_t inputSize, const SValueColum
valCol->offsets.size = cinfo.originalSize; valCol->offsets.size = cinfo.originalSize;
inputStart += cinfo.compressedSize; inputStart += cinfo.compressedSize;
} else { } else {
valCol->numOfValues = info->originalDataSize / tDataTypes[valCol->type].bytes; valCol->numOfValues = info->dataOriginalSize / tDataTypes[valCol->type].bytes;
} }
// data // data
SCompressInfo cinfo = { SCompressInfo cinfo = {
.dataType = valCol->type, .dataType = valCol->type,
.cmprAlg = info->cmprAlg, .cmprAlg = info->cmprAlg,
.originalSize = info->originalDataSize, .originalSize = info->dataOriginalSize,
.compressedSize = info->compressedDataSize, .compressedSize = info->dataCompressedSize,
}; };
code = tBufferEnsureCapacity(&valCol->data, cinfo.originalSize); code = tBufferEnsureCapacity(&valCol->data, cinfo.originalSize);
@ -4226,13 +4199,13 @@ int32_t tValueColumnCompressInfoEncode(const SValueColumnCompressInfo *compressI
if (code) return code; if (code) return code;
code = tBufferPutI8(writer, compressInfo->type); code = tBufferPutI8(writer, compressInfo->type);
if (code) return code; if (code) return code;
code = tBufferPutI32v(writer, compressInfo->originalDataSize); code = tBufferPutI32v(writer, compressInfo->dataOriginalSize);
if (code) return code; if (code) return code;
code = tBufferPutI32v(writer, compressInfo->compressedDataSize); code = tBufferPutI32v(writer, compressInfo->dataCompressedSize);
if (code) return code; if (code) return code;
code = tBufferPutI32v(writer, compressInfo->originalOffsetSize); code = tBufferPutI32v(writer, compressInfo->offsetOriginalSize);
if (code) return code; if (code) return code;
code = tBufferPutI32v(writer, compressInfo->compressedOffsetSize); code = tBufferPutI32v(writer, compressInfo->offsetCompressedSize);
if (code) return code; if (code) return code;
return 0; return 0;
@ -4251,13 +4224,13 @@ int32_t tValueColumnCompressInfoDecode(SBufferReader *reader, SValueColumnCompre
if (code) return code; if (code) return code;
code = tBufferGetI8(reader, &compressInfo->type); code = tBufferGetI8(reader, &compressInfo->type);
if (code) return code; if (code) return code;
code = tBufferGetI32v(reader, &compressInfo->originalDataSize); code = tBufferGetI32v(reader, &compressInfo->dataOriginalSize);
if (code) return code; if (code) return code;
code = tBufferGetI32v(reader, &compressInfo->compressedDataSize); code = tBufferGetI32v(reader, &compressInfo->dataCompressedSize);
if (code) return code; if (code) return code;
code = tBufferGetI32v(reader, &compressInfo->originalOffsetSize); code = tBufferGetI32v(reader, &compressInfo->offsetOriginalSize);
if (code) return code; if (code) return code;
code = tBufferGetI32v(reader, &compressInfo->compressedOffsetSize); code = tBufferGetI32v(reader, &compressInfo->offsetCompressedSize);
if (code) return code; if (code) return code;
} else { } else {
return TSDB_CODE_INVALID_DATA_FMT; return TSDB_CODE_INVALID_DATA_FMT;
@ -4287,13 +4260,16 @@ int32_t tCompressData(void *input, // input
SBuffer local; SBuffer local;
tBufferInit(&local); tBufferInit(&local);
if (info->cmprAlg == TWO_STAGE_COMP) {
if (buffer == NULL) { if (buffer == NULL) {
buffer = &local; buffer = &local;
} }
if (info->cmprAlg == TWO_STAGE_COMP) {
code = tBufferEnsureCapacity(buffer, extraSizeNeeded); code = tBufferEnsureCapacity(buffer, extraSizeNeeded);
if (code) return code; if (code) {
tBufferDestroy(&local);
return code;
}
} }
info->compressedSize = tDataTypes[info->dataType].compFunc( // info->compressedSize = tDataTypes[info->dataType].compFunc( //
@ -4304,7 +4280,7 @@ int32_t tCompressData(void *input, // input
outputSize, // output size outputSize, // output size
info->cmprAlg, // compression algorithm info->cmprAlg, // compression algorithm
buffer->data, // buffer buffer->data, // buffer
extraSizeNeeded // buffer size buffer->capacity // buffer size
); );
if (info->compressedSize < 0) { if (info->compressedSize < 0) {
tBufferDestroy(&local); tBufferDestroy(&local);
@ -4335,12 +4311,16 @@ int32_t tDecompressData(void *input, // input
SBuffer local; SBuffer local;
tBufferInit(&local); tBufferInit(&local);
if (info->cmprAlg == TWO_STAGE_COMP) {
if (buffer == NULL) { if (buffer == NULL) {
buffer = &local; buffer = &local;
} }
if (info->cmprAlg == TWO_STAGE_COMP) {
code = tBufferEnsureCapacity(buffer, info->originalSize + COMP_OVERFLOW_BYTES); code = tBufferEnsureCapacity(buffer, info->originalSize + COMP_OVERFLOW_BYTES);
if (code) return code; if (code) {
tBufferDestroy(&local);
return code;
}
} }
int32_t decompressedSize = tDataTypes[info->dataType].decompFunc( int32_t decompressedSize = tDataTypes[info->dataType].decompFunc(
@ -4378,8 +4358,7 @@ int32_t tCompressDataToBuffer(void *input, int32_t inputSize, SCompressInfo *inf
return 0; return 0;
} }
int32_t tDecompressDataToBuffer(void *input, int32_t inputSize, const SCompressInfo *info, SBuffer *output, int32_t tDecompressDataToBuffer(void *input, int32_t inputSize, SCompressInfo *info, SBuffer *output, SBuffer *assist) {
SBuffer *assist) {
int32_t code; int32_t code;
code = tBufferEnsureCapacity(output, output->size + info->originalSize); code = tBufferEnsureCapacity(output, output->size + info->originalSize);

12
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -23,16 +23,16 @@ static int32_t tsdbFSToBinary(uint8_t *p, STsdbFS *pFS) {
uint32_t nSet = taosArrayGetSize(pFS->aDFileSet); uint32_t nSet = taosArrayGetSize(pFS->aDFileSet);
// version // version
n += tPutI8(p ? p + n : p, 0, true); n += tPutI8(p ? p + n : p, 0);
// SDelFile // SDelFile
n += tPutI8(p ? p + n : p, hasDel, true); n += tPutI8(p ? p + n : p, hasDel);
if (hasDel) { if (hasDel) {
n += tPutDelFile(p ? p + n : p, pFS->pDelFile); n += tPutDelFile(p ? p + n : p, pFS->pDelFile);
} }
// SArray<SDFileSet> // SArray<SDFileSet>
n += tPutU32v(p ? p + n : p, nSet, true); n += tPutU32v(p ? p + n : p, nSet);
for (uint32_t iSet = 0; iSet < nSet; iSet++) { for (uint32_t iSet = 0; iSet < nSet; iSet++) {
n += tPutDFileSet(p ? p + n : p, (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet)); n += tPutDFileSet(p ? p + n : p, (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet));
} }
@ -45,11 +45,11 @@ static int32_t tsdbBinaryToFS(uint8_t *pData, int64_t nData, STsdbFS *pFS) {
int32_t n = 0; int32_t n = 0;
// version // version
n += tGetI8(pData + n, NULL, true); n += tGetI8(pData + n, NULL);
// SDelFile // SDelFile
int8_t hasDel = 0; int8_t hasDel = 0;
n += tGetI8(pData + n, &hasDel, true); n += tGetI8(pData + n, &hasDel);
if (hasDel) { if (hasDel) {
pFS->pDelFile = (SDelFile *)taosMemoryCalloc(1, sizeof(SDelFile)); pFS->pDelFile = (SDelFile *)taosMemoryCalloc(1, sizeof(SDelFile));
if (pFS->pDelFile == NULL) { if (pFS->pDelFile == NULL) {
@ -66,7 +66,7 @@ static int32_t tsdbBinaryToFS(uint8_t *pData, int64_t nData, STsdbFS *pFS) {
// aDFileSet // aDFileSet
taosArrayClear(pFS->aDFileSet); taosArrayClear(pFS->aDFileSet);
uint32_t nSet = 0; uint32_t nSet = 0;
n += tGetU32v(pData + n, &nSet, true); n += tGetU32v(pData + n, &nSet);
for (uint32_t iSet = 0; iSet < nSet; iSet++) { for (uint32_t iSet = 0; iSet < nSet; iSet++) {
SDFileSet fSet = {0}; SDFileSet fSet = {0};

68
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -19,9 +19,9 @@
int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) { int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
int32_t n = 0; int32_t n = 0;
n += tPutI64v(p ? p + n : p, pHeadFile->commitID, true); n += tPutI64v(p ? p + n : p, pHeadFile->commitID);
n += tPutI64v(p ? p + n : p, pHeadFile->size, true); n += tPutI64v(p ? p + n : p, pHeadFile->size);
n += tPutI64v(p ? p + n : p, pHeadFile->offset, true); n += tPutI64v(p ? p + n : p, pHeadFile->offset);
return n; return n;
} }
@ -29,9 +29,9 @@ int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
static int32_t tGetHeadFile(uint8_t *p, SHeadFile *pHeadFile) { static int32_t tGetHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pHeadFile->commitID, true); n += tGetI64v(p + n, &pHeadFile->commitID);
n += tGetI64v(p + n, &pHeadFile->size, true); n += tGetI64v(p + n, &pHeadFile->size);
n += tGetI64v(p + n, &pHeadFile->offset, true); n += tGetI64v(p + n, &pHeadFile->offset);
return n; return n;
} }
@ -39,8 +39,8 @@ static int32_t tGetHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
int32_t tPutDataFile(uint8_t *p, SDataFile *pDataFile) { int32_t tPutDataFile(uint8_t *p, SDataFile *pDataFile) {
int32_t n = 0; int32_t n = 0;
n += tPutI64v(p ? p + n : p, pDataFile->commitID, true); n += tPutI64v(p ? p + n : p, pDataFile->commitID);
n += tPutI64v(p ? p + n : p, pDataFile->size, true); n += tPutI64v(p ? p + n : p, pDataFile->size);
return n; return n;
} }
@ -48,8 +48,8 @@ int32_t tPutDataFile(uint8_t *p, SDataFile *pDataFile) {
static int32_t tGetDataFile(uint8_t *p, SDataFile *pDataFile) { static int32_t tGetDataFile(uint8_t *p, SDataFile *pDataFile) {
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pDataFile->commitID, true); n += tGetI64v(p + n, &pDataFile->commitID);
n += tGetI64v(p + n, &pDataFile->size, true); n += tGetI64v(p + n, &pDataFile->size);
return n; return n;
} }
@ -57,9 +57,9 @@ static int32_t tGetDataFile(uint8_t *p, SDataFile *pDataFile) {
int32_t tPutSttFile(uint8_t *p, SSttFile *pSttFile) { int32_t tPutSttFile(uint8_t *p, SSttFile *pSttFile) {
int32_t n = 0; int32_t n = 0;
n += tPutI64v(p ? p + n : p, pSttFile->commitID, true); n += tPutI64v(p ? p + n : p, pSttFile->commitID);
n += tPutI64v(p ? p + n : p, pSttFile->size, true); n += tPutI64v(p ? p + n : p, pSttFile->size);
n += tPutI64v(p ? p + n : p, pSttFile->offset, true); n += tPutI64v(p ? p + n : p, pSttFile->offset);
return n; return n;
} }
@ -67,9 +67,9 @@ int32_t tPutSttFile(uint8_t *p, SSttFile *pSttFile) {
static int32_t tGetSttFile(uint8_t *p, SSttFile *pSttFile) { static int32_t tGetSttFile(uint8_t *p, SSttFile *pSttFile) {
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pSttFile->commitID, true); n += tGetI64v(p + n, &pSttFile->commitID);
n += tGetI64v(p + n, &pSttFile->size, true); n += tGetI64v(p + n, &pSttFile->size);
n += tGetI64v(p + n, &pSttFile->offset, true); n += tGetI64v(p + n, &pSttFile->offset);
return n; return n;
} }
@ -77,8 +77,8 @@ static int32_t tGetSttFile(uint8_t *p, SSttFile *pSttFile) {
int32_t tPutSmaFile(uint8_t *p, SSmaFile *pSmaFile) { int32_t tPutSmaFile(uint8_t *p, SSmaFile *pSmaFile) {
int32_t n = 0; int32_t n = 0;
n += tPutI64v(p ? p + n : p, pSmaFile->commitID, true); n += tPutI64v(p ? p + n : p, pSmaFile->commitID);
n += tPutI64v(p ? p + n : p, pSmaFile->size, true); n += tPutI64v(p ? p + n : p, pSmaFile->size);
return n; return n;
} }
@ -86,8 +86,8 @@ int32_t tPutSmaFile(uint8_t *p, SSmaFile *pSmaFile) {
static int32_t tGetSmaFile(uint8_t *p, SSmaFile *pSmaFile) { static int32_t tGetSmaFile(uint8_t *p, SSmaFile *pSmaFile) {
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pSmaFile->commitID, true); n += tGetI64v(p + n, &pSmaFile->commitID);
n += tGetI64v(p + n, &pSmaFile->size, true); n += tGetI64v(p + n, &pSmaFile->size);
return n; return n;
} }
@ -221,9 +221,9 @@ _err:
int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) { int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) {
int32_t n = 0; int32_t n = 0;
n += tPutI32v(p ? p + n : p, pSet->diskId.level, true); n += tPutI32v(p ? p + n : p, pSet->diskId.level);
n += tPutI32v(p ? p + n : p, pSet->diskId.id, true); n += tPutI32v(p ? p + n : p, pSet->diskId.id);
n += tPutI32v(p ? p + n : p, pSet->fid, true); n += tPutI32v(p ? p + n : p, pSet->fid);
// data // data
n += tPutHeadFile(p ? p + n : p, pSet->pHeadF); n += tPutHeadFile(p ? p + n : p, pSet->pHeadF);
@ -231,7 +231,7 @@ int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) {
n += tPutSmaFile(p ? p + n : p, pSet->pSmaF); n += tPutSmaFile(p ? p + n : p, pSet->pSmaF);
// stt // stt
n += tPutU8(p ? p + n : p, pSet->nSttF, true); n += tPutU8(p ? p + n : p, pSet->nSttF);
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) { for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
n += tPutSttFile(p ? p + n : p, pSet->aSttF[iStt]); n += tPutSttFile(p ? p + n : p, pSet->aSttF[iStt]);
} }
@ -242,9 +242,9 @@ int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) {
int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) { int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
int32_t n = 0; int32_t n = 0;
n += tGetI32v(p + n, &pSet->diskId.level, true); n += tGetI32v(p + n, &pSet->diskId.level);
n += tGetI32v(p + n, &pSet->diskId.id, true); n += tGetI32v(p + n, &pSet->diskId.id);
n += tGetI32v(p + n, &pSet->fid, true); n += tGetI32v(p + n, &pSet->fid);
// head // head
pSet->pHeadF = (SHeadFile *)taosMemoryCalloc(1, sizeof(SHeadFile)); pSet->pHeadF = (SHeadFile *)taosMemoryCalloc(1, sizeof(SHeadFile));
@ -271,7 +271,7 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
n += tGetSmaFile(p + n, pSet->pSmaF); n += tGetSmaFile(p + n, pSet->pSmaF);
// stt // stt
n += tGetU8(p + n, &pSet->nSttF, true); n += tGetU8(p + n, &pSet->nSttF);
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) { for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
pSet->aSttF[iStt] = (SSttFile *)taosMemoryCalloc(1, sizeof(SSttFile)); pSet->aSttF[iStt] = (SSttFile *)taosMemoryCalloc(1, sizeof(SSttFile));
if (pSet->aSttF[iStt] == NULL) { if (pSet->aSttF[iStt] == NULL) {
@ -298,9 +298,9 @@ void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) {
int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) { int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) {
int32_t n = 0; int32_t n = 0;
n += tPutI64v(p ? p + n : p, pDelFile->commitID, true); n += tPutI64v(p ? p + n : p, pDelFile->commitID);
n += tPutI64v(p ? p + n : p, pDelFile->size, true); n += tPutI64v(p ? p + n : p, pDelFile->size);
n += tPutI64v(p ? p + n : p, pDelFile->offset, true); n += tPutI64v(p ? p + n : p, pDelFile->offset);
return n; return n;
} }
@ -308,9 +308,9 @@ int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) {
int32_t tGetDelFile(uint8_t *p, SDelFile *pDelFile) { int32_t tGetDelFile(uint8_t *p, SDelFile *pDelFile) {
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pDelFile->commitID, true); n += tGetI64v(p + n, &pDelFile->commitID);
n += tGetI64v(p + n, &pDelFile->size, true); n += tGetI64v(p + n, &pDelFile->size);
n += tGetI64v(p + n, &pDelFile->offset, true); n += tGetI64v(p + n, &pDelFile->offset);
return n; return n;
} }

8
source/dnode/vnode/src/tsdb/tsdbSttFileRW.c
View File

@ -522,17 +522,17 @@ int32_t tsdbSttFileReadStatisBlock(SSttFileReader *reader, const SStatisBlk *sta
// decode value columns // decode value columns
for (int32_t i = 0; i < statisBlk->numOfPKs; i++) { for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
code = tValueColumnDecompress(tBufferGetDataAt(&reader->buffers[0], size), firstKeyInfos[i].compressedDataSize, code = tValueColumnDecompress(tBufferGetDataAt(&reader->buffers[0], size), firstKeyInfos[i].dataCompressedSize,
&firstKeyInfos[i], &statisBlock->firstKeyPKs[i], &reader->buffers[1]); &firstKeyInfos[i], &statisBlock->firstKeyPKs[i], &reader->buffers[1]);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
size += firstKeyInfos[i].compressedDataSize; size += firstKeyInfos[i].dataCompressedSize;
} }
for (int32_t i = 0; i < statisBlk->numOfPKs; i++) { for (int32_t i = 0; i < statisBlk->numOfPKs; i++) {
code = tValueColumnDecompress(tBufferGetDataAt(&reader->buffers[0], size), lastKeyInfos[i].compressedDataSize, code = tValueColumnDecompress(tBufferGetDataAt(&reader->buffers[0], size), lastKeyInfos[i].dataCompressedSize,
&lastKeyInfos[i], &statisBlock->lastKeyPKs[i], &reader->buffers[1]); &lastKeyInfos[i], &statisBlock->lastKeyPKs[i], &reader->buffers[1]);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
size += lastKeyInfos[i].compressedDataSize; size += lastKeyInfos[i].dataCompressedSize;
} }
} }

178
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -151,18 +151,18 @@ int32_t tGetMapData(uint8_t *p, SMapData *pMapData) {
tMapDataReset(pMapData); tMapDataReset(pMapData);
n += tGetI32v(p + n, &pMapData->nItem, true); n += tGetI32v(p + n, &pMapData->nItem);
if (pMapData->nItem) { if (pMapData->nItem) {
if (tRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem)) return -1; if (tRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem)) return -1;
int32_t lOffset = 0; int32_t lOffset = 0;
for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) { for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) {
n += tGetI32v(p + n, &pMapData->aOffset[iItem], true); n += tGetI32v(p + n, &pMapData->aOffset[iItem]);
pMapData->aOffset[iItem] += lOffset; pMapData->aOffset[iItem] += lOffset;
lOffset = pMapData->aOffset[iItem]; lOffset = pMapData->aOffset[iItem];
} }
n += tGetI32v(p + n, &pMapData->nData, true); n += tGetI32v(p + n, &pMapData->nData);
if (tRealloc(&pMapData->pData, pMapData->nData)) return -1; if (tRealloc(&pMapData->pData, pMapData->nData)) return -1;
memcpy(pMapData->pData, p + n, pMapData->nData); memcpy(pMapData->pData, p + n, pMapData->nData);
n += pMapData->nData; n += pMapData->nData;
@ -210,10 +210,10 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SBlockIdx *pBlockIdx = (SBlockIdx *)ph; SBlockIdx *pBlockIdx = (SBlockIdx *)ph;
n += tGetI64(p + n, &pBlockIdx->suid, true); n += tGetI64(p + n, &pBlockIdx->suid);
n += tGetI64(p + n, &pBlockIdx->uid, true); n += tGetI64(p + n, &pBlockIdx->uid);
n += tGetI64v(p + n, &pBlockIdx->offset, true); n += tGetI64v(p + n, &pBlockIdx->offset);
n += tGetI64v(p + n, &pBlockIdx->size, true); n += tGetI64v(p + n, &pBlockIdx->size);
return n; return n;
} }
@ -293,23 +293,23 @@ int32_t tGetDataBlk(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SDataBlk *pDataBlk = (SDataBlk *)ph; SDataBlk *pDataBlk = (SDataBlk *)ph;
n += tGetI64v(p + n, &pDataBlk->minKey.version, true); n += tGetI64v(p + n, &pDataBlk->minKey.version);
n += tGetI64v(p + n, &pDataBlk->minKey.ts, true); n += tGetI64v(p + n, &pDataBlk->minKey.ts);
n += tGetI64v(p + n, &pDataBlk->maxKey.version, true); n += tGetI64v(p + n, &pDataBlk->maxKey.version);
n += tGetI64v(p + n, &pDataBlk->maxKey.ts, true); n += tGetI64v(p + n, &pDataBlk->maxKey.ts);
n += tGetI64v(p + n, &pDataBlk->minVer, true); n += tGetI64v(p + n, &pDataBlk->minVer);
n += tGetI64v(p + n, &pDataBlk->maxVer, true); n += tGetI64v(p + n, &pDataBlk->maxVer);
n += tGetI32v(p + n, &pDataBlk->nRow, true); n += tGetI32v(p + n, &pDataBlk->nRow);
n += tGetI8(p + n, &pDataBlk->hasDup, true); n += tGetI8(p + n, &pDataBlk->hasDup);
n += tGetI8(p + n, &pDataBlk->nSubBlock, true); n += tGetI8(p + n, &pDataBlk->nSubBlock);
for (int8_t iSubBlock = 0; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) { for (int8_t iSubBlock = 0; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
n += tGetI64v(p + n, &pDataBlk->aSubBlock[iSubBlock].offset, true); n += tGetI64v(p + n, &pDataBlk->aSubBlock[iSubBlock].offset);
n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szBlock, true); n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szBlock);
n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szKey, true); n += tGetI32v(p + n, &pDataBlk->aSubBlock[iSubBlock].szKey);
} }
if (pDataBlk->nSubBlock == 1 && !pDataBlk->hasDup) { if (pDataBlk->nSubBlock == 1 && !pDataBlk->hasDup) {
n += tGetI64v(p + n, &pDataBlk->smaInfo.offset, true); n += tGetI64v(p + n, &pDataBlk->smaInfo.offset);
n += tGetI32v(p + n, &pDataBlk->smaInfo.size, true); n += tGetI32v(p + n, &pDataBlk->smaInfo.size);
} else { } else {
pDataBlk->smaInfo.offset = 0; pDataBlk->smaInfo.offset = 0;
pDataBlk->smaInfo.size = 0; pDataBlk->smaInfo.size = 0;
@ -364,17 +364,17 @@ int32_t tGetSttBlk(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SSttBlk *pSttBlk = (SSttBlk *)ph; SSttBlk *pSttBlk = (SSttBlk *)ph;
n += tGetI64(p + n, &pSttBlk->suid, true); n += tGetI64(p + n, &pSttBlk->suid);
n += tGetI64(p + n, &pSttBlk->minUid, true); n += tGetI64(p + n, &pSttBlk->minUid);
n += tGetI64(p + n, &pSttBlk->maxUid, true); n += tGetI64(p + n, &pSttBlk->maxUid);
n += tGetI64v(p + n, &pSttBlk->minKey, true); n += tGetI64v(p + n, &pSttBlk->minKey);
n += tGetI64v(p + n, &pSttBlk->maxKey, true); n += tGetI64v(p + n, &pSttBlk->maxKey);
n += tGetI64v(p + n, &pSttBlk->minVer, true); n += tGetI64v(p + n, &pSttBlk->minVer);
n += tGetI64v(p + n, &pSttBlk->maxVer, true); n += tGetI64v(p + n, &pSttBlk->maxVer);
n += tGetI32v(p + n, &pSttBlk->nRow, true); n += tGetI32v(p + n, &pSttBlk->nRow);
n += tGetI64v(p + n, &pSttBlk->bInfo.offset, true); n += tGetI64v(p + n, &pSttBlk->bInfo.offset);
n += tGetI32v(p + n, &pSttBlk->bInfo.szBlock, true); n += tGetI32v(p + n, &pSttBlk->bInfo.szBlock);
n += tGetI32v(p + n, &pSttBlk->bInfo.szKey, true); n += tGetI32v(p + n, &pSttBlk->bInfo.szKey);
return n; return n;
} }
@ -386,26 +386,26 @@ int32_t tPutBlockCol(uint8_t *p, void *ph) {
ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE)); ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE));
n += tPutI16v(p ? p + n : p, pBlockCol->cid, true); n += tPutI16v(p ? p + n : p, pBlockCol->cid);
n += tPutI8(p ? p + n : p, pBlockCol->type, true); n += tPutI8(p ? p + n : p, pBlockCol->type);
n += tPutI8(p ? p + n : p, pBlockCol->cflag, true); n += tPutI8(p ? p + n : p, pBlockCol->cflag);
n += tPutI8(p ? p + n : p, pBlockCol->flag, true); n += tPutI8(p ? p + n : p, pBlockCol->flag);
n += tPutI32v(p ? p + n : p, pBlockCol->szOrigin, true); n += tPutI32v(p ? p + n : p, pBlockCol->szOrigin);
if (pBlockCol->flag != HAS_NULL) { if (pBlockCol->flag != HAS_NULL) {
if (pBlockCol->flag != HAS_VALUE) { if (pBlockCol->flag != HAS_VALUE) {
n += tPutI32v(p ? p + n : p, pBlockCol->szBitmap, true); n += tPutI32v(p ? p + n : p, pBlockCol->szBitmap);
} }
if (IS_VAR_DATA_TYPE(pBlockCol->type)) { if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
n += tPutI32v(p ? p + n : p, pBlockCol->szOffset, true); n += tPutI32v(p ? p + n : p, pBlockCol->szOffset);
} }
if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) { if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
n += tPutI32v(p ? p + n : p, pBlockCol->szValue, true); n += tPutI32v(p ? p + n : p, pBlockCol->szValue);
} }
n += tPutI32v(p ? p + n : p, pBlockCol->offset, true); n += tPutI32v(p ? p + n : p, pBlockCol->offset);
} }
_exit: _exit:
@ -416,11 +416,11 @@ int32_t tGetBlockCol(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SBlockCol *pBlockCol = (SBlockCol *)ph; SBlockCol *pBlockCol = (SBlockCol *)ph;
n += tGetI16v(p + n, &pBlockCol->cid, true); n += tGetI16v(p + n, &pBlockCol->cid);
n += tGetI8(p + n, &pBlockCol->type, true); n += tGetI8(p + n, &pBlockCol->type);
n += tGetI8(p + n, &pBlockCol->cflag, true); n += tGetI8(p + n, &pBlockCol->cflag);
n += tGetI8(p + n, &pBlockCol->flag, true); n += tGetI8(p + n, &pBlockCol->flag);
n += tGetI32v(p + n, &pBlockCol->szOrigin, true); n += tGetI32v(p + n, &pBlockCol->szOrigin);
ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE)); ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE));
@ -431,18 +431,18 @@ int32_t tGetBlockCol(uint8_t *p, void *ph) {
if (pBlockCol->flag != HAS_NULL) { if (pBlockCol->flag != HAS_NULL) {
if (pBlockCol->flag != HAS_VALUE) { if (pBlockCol->flag != HAS_VALUE) {
n += tGetI32v(p + n, &pBlockCol->szBitmap, true); n += tGetI32v(p + n, &pBlockCol->szBitmap);
} }
if (IS_VAR_DATA_TYPE(pBlockCol->type)) { if (IS_VAR_DATA_TYPE(pBlockCol->type)) {
n += tGetI32v(p + n, &pBlockCol->szOffset, true); n += tGetI32v(p + n, &pBlockCol->szOffset);
} }
if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) { if (pBlockCol->flag != (HAS_NULL | HAS_NONE)) {
n += tGetI32v(p + n, &pBlockCol->szValue, true); n += tGetI32v(p + n, &pBlockCol->szValue);
} }
n += tGetI32v(p + n, &pBlockCol->offset, true); n += tGetI32v(p + n, &pBlockCol->offset);
} }
return n; return n;
@ -496,10 +496,10 @@ int32_t tGetDelIdx(uint8_t *p, void *ph) {
SDelIdx *pDelIdx = (SDelIdx *)ph; SDelIdx *pDelIdx = (SDelIdx *)ph;
int32_t n = 0; int32_t n = 0;
n += tGetI64(p + n, &pDelIdx->suid, true); n += tGetI64(p + n, &pDelIdx->suid);
n += tGetI64(p + n, &pDelIdx->uid, true); n += tGetI64(p + n, &pDelIdx->uid);
n += tGetI64v(p + n, &pDelIdx->offset, true); n += tGetI64v(p + n, &pDelIdx->offset);
n += tGetI64v(p + n, &pDelIdx->size, true); n += tGetI64v(p + n, &pDelIdx->size);
return n; return n;
} }
@ -522,9 +522,9 @@ int32_t tGetDelData(uint8_t *p, void *ph) {
SDelData *pDelData = (SDelData *)ph; SDelData *pDelData = (SDelData *)ph;
int32_t n = 0; int32_t n = 0;
n += tGetI64v(p + n, &pDelData->version, true); n += tGetI64v(p + n, &pDelData->version);
n += tGetI64(p + n, &pDelData->sKey, true); n += tGetI64(p + n, &pDelData->sKey);
n += tGetI64(p + n, &pDelData->eKey, true); n += tGetI64(p + n, &pDelData->eKey);
return n; return n;
} }
@ -1595,18 +1595,18 @@ _exit:
int32_t tPutDiskDataHdr(uint8_t *p, const SDiskDataHdr *pHdr) { int32_t tPutDiskDataHdr(uint8_t *p, const SDiskDataHdr *pHdr) {
int32_t n = 0; int32_t n = 0;
n += tPutU32(p ? p + n : p, pHdr->delimiter, true); n += tPutU32(p ? p + n : p, pHdr->delimiter);
n += tPutU32v(p ? p + n : p, pHdr->fmtVer, true); n += tPutU32v(p ? p + n : p, pHdr->fmtVer);
n += tPutI64(p ? p + n : p, pHdr->suid, true); n += tPutI64(p ? p + n : p, pHdr->suid);
n += tPutI64(p ? p + n : p, pHdr->uid, true); n += tPutI64(p ? p + n : p, pHdr->uid);
n += tPutI32v(p ? p + n : p, pHdr->szUid, true); n += tPutI32v(p ? p + n : p, pHdr->szUid);
n += tPutI32v(p ? p + n : p, pHdr->szVer, true); n += tPutI32v(p ? p + n : p, pHdr->szVer);
n += tPutI32v(p ? p + n : p, pHdr->szKey, true); n += tPutI32v(p ? p + n : p, pHdr->szKey);
n += tPutI32v(p ? p + n : p, pHdr->szBlkCol, true); n += tPutI32v(p ? p + n : p, pHdr->szBlkCol);
n += tPutI32v(p ? p + n : p, pHdr->nRow, true); n += tPutI32v(p ? p + n : p, pHdr->nRow);
n += tPutI8(p ? p + n : p, pHdr->cmprAlg, true); n += tPutI8(p ? p + n : p, pHdr->cmprAlg);
if (pHdr->fmtVer == 1) { if (pHdr->fmtVer == 1) {
n += tPutI8(p ? p + n : p, pHdr->numPrimaryKeyCols, true); n += tPutI8(p ? p + n : p, pHdr->numPrimaryKeyCols);
} }
return n; return n;
@ -1616,18 +1616,18 @@ int32_t tGetDiskDataHdr(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SDiskDataHdr *pHdr = (SDiskDataHdr *)ph; SDiskDataHdr *pHdr = (SDiskDataHdr *)ph;
n += tGetU32(p + n, &pHdr->delimiter, true); n += tGetU32(p + n, &pHdr->delimiter);
n += tGetU32v(p + n, &pHdr->fmtVer, true); n += tGetU32v(p + n, &pHdr->fmtVer);
n += tGetI64(p + n, &pHdr->suid, true); n += tGetI64(p + n, &pHdr->suid);
n += tGetI64(p + n, &pHdr->uid, true); n += tGetI64(p + n, &pHdr->uid);
n += tGetI32v(p + n, &pHdr->szUid, true); n += tGetI32v(p + n, &pHdr->szUid);
n += tGetI32v(p + n, &pHdr->szVer, true); n += tGetI32v(p + n, &pHdr->szVer);
n += tGetI32v(p + n, &pHdr->szKey, true); n += tGetI32v(p + n, &pHdr->szKey);
n += tGetI32v(p + n, &pHdr->szBlkCol, true); n += tGetI32v(p + n, &pHdr->szBlkCol);
n += tGetI32v(p + n, &pHdr->nRow, true); n += tGetI32v(p + n, &pHdr->nRow);
n += tGetI8(p + n, &pHdr->cmprAlg, true); n += tGetI8(p + n, &pHdr->cmprAlg);
if (pHdr->fmtVer == 1) { if (pHdr->fmtVer == 1) {
n += tGetI8(p + n, &pHdr->numPrimaryKeyCols, true); n += tGetI8(p + n, &pHdr->numPrimaryKeyCols);
} else { } else {
pHdr->numPrimaryKeyCols = 0; pHdr->numPrimaryKeyCols = 0;
} }
@ -1639,11 +1639,11 @@ int32_t tGetDiskDataHdr(uint8_t *p, void *ph) {
int32_t tPutColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg) { int32_t tPutColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg) {
int32_t n = 0; int32_t n = 0;
n += tPutI16v(p ? p + n : p, pColAgg->colId, true); n += tPutI16v(p ? p + n : p, pColAgg->colId);
n += tPutI16v(p ? p + n : p, pColAgg->numOfNull, true); n += tPutI16v(p ? p + n : p, pColAgg->numOfNull);
n += tPutI64(p ? p + n : p, pColAgg->sum, true); n += tPutI64(p ? p + n : p, pColAgg->sum);
n += tPutI64(p ? p + n : p, pColAgg->max, true); n += tPutI64(p ? p + n : p, pColAgg->max);
n += tPutI64(p ? p + n : p, pColAgg->min, true); n += tPutI64(p ? p + n : p, pColAgg->min);
return n; return n;
} }
@ -1651,11 +1651,11 @@ int32_t tPutColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg) {
int32_t tGetColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg) { int32_t tGetColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg) {
int32_t n = 0; int32_t n = 0;
n += tGetI16v(p + n, &pColAgg->colId, true); n += tGetI16v(p + n, &pColAgg->colId);
n += tGetI16v(p + n, &pColAgg->numOfNull, true); n += tGetI16v(p + n, &pColAgg->numOfNull);
n += tGetI64(p + n, &pColAgg->sum, true); n += tGetI64(p + n, &pColAgg->sum);
n += tGetI64(p + n, &pColAgg->max, true); n += tGetI64(p + n, &pColAgg->max);
n += tGetI64(p + n, &pColAgg->min, true); n += tGetI64(p + n, &pColAgg->min);
return n; return n;
} }