Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
“4b90cb2b82622ab15662f2fafe525c3fe8f00927”上不存在“docs/en/git@gitcode.net:taosdata/tdengine.git”
未验证 提交 f58396ea 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #18418 from taosdata/revert-18415-revert-18396-enh/row_optimize2 

Revert "Revert "enh: row optimize2""
上级 e590b620 335a572d
隐藏空白更改
内联 并排
Showing with 686 addition and 617 deletion
include/common/tdataformat.h
浏览文件 @ f58396ea
...@@ -27,17 +27,17 @@ ...@@ -27,17 +27,17 @@
extern "C" { extern "C" {
#endif #endif
typedef struct SBuffer SBuffer; typedef struct SBuffer SBuffer;
typedef struct SSchema SSchema; typedef struct SSchema SSchema;
typedef struct STColumn STColumn; typedef struct STColumn STColumn;
typedef struct STSchema STSchema; typedef struct STSchema STSchema;
typedef struct SValue SValue; typedef struct SValue SValue;
typedef struct SColVal SColVal; typedef struct SColVal SColVal;
typedef struct STSRow2 STSRow2; typedef struct SRow SRow;
typedef struct STSRowBuilder STSRowBuilder; typedef struct SRowIter SRowIter;
typedef struct STagVal STagVal; typedef struct STagVal STagVal;
typedef struct STag STag; typedef struct STag STag;
typedef struct SColData SColData; typedef struct SColData SColData;
#define HAS_NONE ((uint8_t)0x1) #define HAS_NONE ((uint8_t)0x1)
#define HAS_NULL ((uint8_t)0x2) #define HAS_NULL ((uint8_t)0x2)
...@@ -68,13 +68,10 @@ struct SBuffer { ...@@ -68,13 +68,10 @@ struct SBuffer {
void tBufferDestroy(SBuffer *pBuffer); void tBufferDestroy(SBuffer *pBuffer);
int32_t tBufferInit(SBuffer *pBuffer, int64_t size); int32_t tBufferInit(SBuffer *pBuffer, int64_t size);
int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData); int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData);
int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
// STSchema ================================ // STSchema ================================
int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema); void tDestroyTSchema(STSchema *pTSchema);
void tTSchemaDestroy(STSchema *pTSchema);
// SValue ================================
static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type);
// SColVal ================================ // SColVal ================================
#define CV_FLAG_VALUE ((int8_t)0x0) #define CV_FLAG_VALUE ((int8_t)0x0)
...@@ -89,26 +86,14 @@ static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type); ...@@ -89,26 +86,14 @@ static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type);
#define COL_VAL_IS_NULL(CV) ((CV)->flag == CV_FLAG_NULL) #define COL_VAL_IS_NULL(CV) ((CV)->flag == CV_FLAG_NULL)
#define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE) #define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
// STSRow2 ================================ // SRow ================================
#define TSROW_LEN(PROW, V) tGetI32v((uint8_t *)(PROW)->data, (V) ? &(V) : NULL) int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SBuffer *pBuffer);
#define TSROW_SVER(PROW, V) tGetI32v((PROW)->data + TSROW_LEN(PROW, NULL), (V) ? &(V) : NULL) void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow); // SRowIter ================================
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow); int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter);
void tTSRowFree(STSRow2 *pRow); void tRowIterClose(SRowIter **ppIter);
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal); SColVal *tRowIterNext(SRowIter *pIter);
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray);
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
int32_t tGetTSRow(uint8_t *p, STSRow2 **ppRow);
// STSRowBuilder ================================
#define tsRowBuilderInit() ((STSRowBuilder){0})
#define tsRowBuilderClear(B) \
do { \
if ((B)->pBuf) { \
taosMemoryFree((B)->pBuf); \
} \
} while (0)
// STag ================================ // STag ================================
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag); int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
...@@ -147,29 +132,17 @@ struct STSchema { ...@@ -147,29 +132,17 @@ struct STSchema {
int32_t numOfCols; int32_t numOfCols;
int32_t version; int32_t version;
int32_t flen; int32_t flen;
int32_t vlen;
int32_t tlen; int32_t tlen;
STColumn columns[]; STColumn columns[];
}; };
#define TSROW_HAS_NONE ((uint8_t)0x1) struct SRow {
#define TSROW_HAS_NULL ((uint8_t)0x2U) uint8_t flag;
#define TSROW_HAS_VAL ((uint8_t)0x4U) uint8_t rsv;
#define TSROW_KV_SMALL ((uint8_t)0x10U) uint16_t sver;
#define TSROW_KV_MID ((uint8_t)0x20U) uint32_t len;
#define TSROW_KV_BIG ((uint8_t)0x40U) TSKEY ts;
#pragma pack(push, 1) uint8_t data[];
struct STSRow2 {
TSKEY ts;
uint8_t flags;
uint8_t data[];
};
#pragma pack(pop)
struct STSRowBuilder {
// STSRow2 tsRow;
int32_t szBuf;
uint8_t *pBuf;
}; };
struct SValue { struct SValue {
...@@ -258,37 +231,17 @@ typedef struct { ...@@ -258,37 +231,17 @@ typedef struct {
int32_t nCols; int32_t nCols;
schema_ver_t version; schema_ver_t version;
uint16_t flen; uint16_t flen;
int32_t vlen;
int32_t tlen; int32_t tlen;
STColumn *columns; STColumn *columns;
} STSchemaBuilder; } STSchemaBuilder;
// use 2 bits for bitmap(default: STSRow/sub block)
#define TD_VTYPE_BITS 2
#define TD_VTYPE_PARTS 4 // PARTITIONS: 1 byte / 2 bits
#define TD_VTYPE_OPTR 3 // OPERATOR: 4 - 1, utilize to get remainder
#define TD_BITMAP_BYTES(cnt) (((cnt) + TD_VTYPE_OPTR) >> 2)
// use 1 bit for bitmap(super block)
#define TD_VTYPE_BITS_I 1
#define TD_VTYPE_PARTS_I 8 // PARTITIONS: 1 byte / 1 bit
#define TD_VTYPE_OPTR_I 7 // OPERATOR: 8 - 1, utilize to get remainder
#define TD_BITMAP_BYTES_I(cnt) (((cnt) + TD_VTYPE_OPTR_I) >> 3)
int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version); int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder); void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder);
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version); void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, col_id_t colId, col_bytes_t bytes); int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, col_id_t colId, col_bytes_t bytes);
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder); STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) { STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version);
if (IS_VAR_DATA_TYPE(type)) {
return tGetBinary(p, &pValue->pData, pValue ? &pValue->nData : NULL);
} else {
memcpy(&pValue->val, p, tDataTypes[type].bytes);
return tDataTypes[type].bytes;
}
}
#endif #endif
......
include/common/trow.h
浏览文件 @ f58396ea
...@@ -55,6 +55,14 @@ typedef struct STSRow { ...@@ -55,6 +55,14 @@ typedef struct STSRow {
#define TD_ROW_TP 0x0U // default #define TD_ROW_TP 0x0U // default
#define TD_ROW_KV 0x01U #define TD_ROW_KV 0x01U
#define TD_VTYPE_PARTS 4 // PARTITIONS: 1 byte / 2 bits
#define TD_VTYPE_OPTR 3 // OPERATOR: 4 - 1, utilize to get remainder
#define TD_BITMAP_BYTES(cnt) (((cnt) + TD_VTYPE_OPTR) >> 2)
#define TD_VTYPE_PARTS_I 8 // PARTITIONS: 1 byte / 1 bit
#define TD_VTYPE_OPTR_I 7 // OPERATOR: 8 - 1, utilize to get remainder
#define TD_BITMAP_BYTES_I(cnt) (((cnt) + TD_VTYPE_OPTR_I) >> 3)
/** /**
* @brief value type * @brief value type
* - for data from client input and STSRow in memory, 3 types of value none/null/norm available * - for data from client input and STSRow in memory, 3 types of value none/null/norm available
...@@ -244,7 +252,7 @@ int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pVal ...@@ -244,7 +252,7 @@ int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pVal
*/ */
static FORCE_INLINE void *tdGetBitmapAddrTp(STSRow *pRow, uint32_t flen) { static FORCE_INLINE void *tdGetBitmapAddrTp(STSRow *pRow, uint32_t flen) {
// The primary TS key is stored separatedly. // The primary TS key is stored separatedly.
return POINTER_SHIFT(TD_ROW_DATA(pRow), flen - sizeof(TSKEY)); return POINTER_SHIFT(TD_ROW_DATA(pRow), flen);
// return POINTER_SHIFT(pRow->ts, flen); // return POINTER_SHIFT(pRow->ts, flen);
} }
......
source/client/src/clientRawBlockWrite.c
浏览文件 @ f58396ea
...@@ -1275,6 +1275,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) ...@@ -1275,6 +1275,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
nVar++; nVar++;
} }
} }
fLen -= sizeof(TSKEY);
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) + int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
(int32_t)TD_BITMAP_BYTES(numOfCols - 1); (int32_t)TD_BITMAP_BYTES(numOfCols - 1);
...@@ -1333,7 +1334,9 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) ...@@ -1333,7 +1334,9 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
} }
} }
offset += TYPE_BYTES[pColumn->type]; if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type];
}
} }
tdSRowEnd(&rb); tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData); int32_t rowLen = TD_ROW_LEN(rowData);
...@@ -1503,6 +1506,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { ...@@ -1503,6 +1506,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
nVar++; nVar++;
} }
} }
fLen -= sizeof(TSKEY);
int32_t rows = rspObj.resInfo.numOfRows; int32_t rows = rspObj.resInfo.numOfRows;
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) + int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
...@@ -1585,8 +1589,9 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { ...@@ -1585,8 +1589,9 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k); tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k);
} }
} }
if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type]; offset += TYPE_BYTES[pColumn->type];
}
} }
tdSRowEnd(&rb); tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData); int32_t rowLen = TD_ROW_LEN(rowData);
...@@ -1803,6 +1808,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) ...@@ -1803,6 +1808,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
nVar++; nVar++;
} }
} }
fLen -= sizeof(TSKEY);
int32_t rows = rspObj.resInfo.numOfRows; int32_t rows = rspObj.resInfo.numOfRows;
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) + int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
...@@ -1888,8 +1894,9 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) ...@@ -1888,8 +1894,9 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k); tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k);
} }
} }
if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type]; offset += TYPE_BYTES[pColumn->type];
}
} }
tdSRowEnd(&rb); tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData); int32_t rowLen = TD_ROW_LEN(rowData);
......
source/common/src/tdatablock.c
浏览文件 @ f58396ea
...@@ -2055,6 +2055,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB ...@@ -2055,6 +2055,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
isStartKey = true; isStartKey = true;
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true, tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true,
offset, k); offset, k);
continue; // offset should keep 0 for next column
} else if (colDataIsNull_s(pColInfoData, j)) { } else if (colDataIsNull_s(pColInfoData, j)) {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NULL, NULL, tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NULL, NULL,
......
source/common/src/tdataformat.c
浏览文件 @ f58396ea
...@@ -43,606 +43,672 @@ int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData) { ...@@ -43,606 +43,672 @@ int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData) {
return code; return code;
} }
int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData) {
int32_t code = tRealloc(&pBuffer->pBuf, pBuffer->nBuf + nData);
if (code) return code;
*ppData = pBuffer->pBuf + pBuffer->nBuf;
pBuffer->nBuf += nData;
return code;
}
// ================================ // ================================
static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson); static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson);
// SRow ========================================================================
#define KV_FLG_LIT ((uint8_t)0x10)
#define KV_FLG_MID ((uint8_t)0x20)
#define KV_FLG_BIG ((uint8_t)0x30)
#define ROW_BIT_NONE ((uint8_t)0x0)
#define ROW_BIT_NULL ((uint8_t)0x1)
#define ROW_BIT_VALUE ((uint8_t)0x2)
#pragma pack(push, 1) #pragma pack(push, 1)
typedef struct { typedef struct {
int16_t nCols; int16_t nCol;
uint8_t idx[]; char idx[]; // uint8_t * | uint16_t * | uint32_t *
} STSKVRow; } SKVIdx;
#pragma pack(pop) #pragma pack(pop)
#define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW) #define ROW_SET_BITMAP(PB, FLAG, IDX, VAL) \
do { \
// SValue if (PB) { \
static FORCE_INLINE int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) { switch (FLAG) { \
if (IS_VAR_DATA_TYPE(type)) { case (HAS_NULL | HAS_NONE): \
return tPutBinary(p, pValue->pData, pValue->nData); SET_BIT1(PB, IDX, VAL); \
} else { break; \
if (p) memcpy(p, &pValue->val, tDataTypes[type].bytes); case (HAS_VALUE | HAS_NONE): \
return tDataTypes[type].bytes; SET_BIT1(PB, IDX, (VAL) ? (VAL)-1 : 0); \
} break; \
} case (HAS_VALUE | HAS_NULL): \
SET_BIT1(PB, IDX, (VAL)-1); \
// STSRow2 ======================================================================== break; \
static void setBitMap(uint8_t *pb, uint8_t v, int32_t idx, uint8_t flags) { case (HAS_VALUE | HAS_NULL | HAS_NONE): \
if (pb) { SET_BIT2(PB, IDX, VAL); \
switch (flags & 0xf) { break; \
case TSROW_HAS_NULL | TSROW_HAS_NONE: default: \
case TSROW_HAS_VAL | TSROW_HAS_NONE: ASSERT(0); \
if (v) { break; \
SET_BIT1(pb, idx, (uint8_t)1); } \
} else { } \
SET_BIT1(pb, idx, (uint8_t)0);
}
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL:
v = v - 1;
SET_BIT1(pb, idx, v);
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
SET_BIT2(pb, idx, v);
break;
default:
ASSERT(0);
}
}
}
#define SET_IDX(p, i, n, f) \
do { \
if ((f)&TSROW_KV_SMALL) { \
((uint8_t *)(p))[i] = (n); \
} else if ((f)&TSROW_KV_MID) { \
((uint16_t *)(p))[i] = (n); \
} else { \
((uint32_t *)(p))[i] = (n); \
} \
} while (0) } while (0)
#if 0
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
int32_t code = 0;
#if 0
STColumn *pTColumn;
SColVal *pColVal;
int32_t nColVal = taosArrayGetSize(pArray);
int32_t iColVal;
ASSERT(nColVal > 0);
// try
uint8_t flags = 0;
uint32_t ntv = 0;
uint32_t nkv = 0;
int16_t nTag = 0;
uint32_t maxIdx = 0;
iColVal = 0;
for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; iColumn++) {
pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else {
pColVal = NULL;
}
if (iColumn == 0) { int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SBuffer *pBuffer) {
ASSERT(pColVal->cid == pTColumn->colId); int32_t code = 0;
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
ASSERT(pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
iColVal++;
} else {
if (pColVal) {
if (pColVal->cid == pTColumn->colId) {
iColVal++;
if (COL_VAL_IS_NONE(pColVal)) { ASSERT(taosArrayGetSize(aColVal) > 0);
flags |= TSROW_HAS_NONE; ASSERT(((SColVal *)aColVal->pData)[0].cid == PRIMARYKEY_TIMESTAMP_COL_ID);
} else if (COL_VAL_IS_NULL(pColVal)) { ASSERT(((SColVal *)aColVal->pData)[0].type == TSDB_DATA_TYPE_TIMESTAMP);
flags |= TSROW_HAS_NULL;
maxIdx = nkv; // scan ---------------
nTag++; uint8_t flag = 0;
nkv += tPutI16v(NULL, -pTColumn->colId); int32_t iColVal = 1;
const int32_t nColVal = taosArrayGetSize(aColVal);
SColVal *pColVal = (iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
int32_t iTColumn = 1;
STColumn *pTColumn = pTSchema->columns + iTColumn;
int32_t ntp = 0;
int32_t nkv = 0;
int32_t maxIdx = 0;
int32_t nIdx = 0;
while (pTColumn) {
if (pColVal) {
if (pColVal->cid == pTColumn->colId) {
ntp += TYPE_BYTES[pTColumn->type];
if (COL_VAL_IS_VALUE(pColVal)) { // VALUE
flag |= HAS_VALUE;
maxIdx = nkv;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ntp = ntp + tPutU32v(NULL, pColVal->value.nData) + pColVal->value.nData;
nkv = nkv + tPutI16v(NULL, pTColumn->colId) + tPutU32v(NULL, pColVal->value.nData) + pColVal->value.nData;
} else { } else {
flags |= TSROW_HAS_VAL; nkv = nkv + tPutI16v(NULL, pTColumn->colId) + pTColumn->bytes;
maxIdx = nkv;
nTag++;
nkv += tPutI16v(NULL, pTColumn->colId);
nkv += tPutValue(NULL, &pColVal->value, pTColumn->type);
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ntv += tPutValue(NULL, &pColVal->value, pTColumn->type);
}
} }
} else if (pColVal->cid > pTColumn->colId) { nIdx++;
flags |= TSROW_HAS_NONE; } else if (COL_VAL_IS_NONE(pColVal)) { // NONE
flag |= HAS_NONE;
} else if (COL_VAL_IS_NULL(pColVal)) { // NULL
flag |= HAS_NULL;
maxIdx = nkv;
nkv += tPutI16v(NULL, -pTColumn->colId);
nIdx++;
} else { } else {
ASSERT(0); ASSERT(0);
} }
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
} else if (pColVal->cid > pTColumn->colId) { // NONE
flag |= HAS_NONE;
ntp += TYPE_BYTES[pTColumn->type];
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
} else { } else {
flags |= TSROW_HAS_NONE; pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
} }
} else { // NONE
flag |= HAS_NONE;
ntp += TYPE_BYTES[pTColumn->type];
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
} }
} }
ASSERT(flags); // compare ---------------
switch (flag) {
// decide case HAS_NONE:
uint32_t nData = 0; case HAS_NULL:
uint32_t nDataT = 0; ntp = sizeof(SRow);
uint32_t nDataK = 0; break;
if (flags == TSROW_HAS_NONE || flags == TSROW_HAS_NULL) { case HAS_VALUE:
nData = 0; ntp = sizeof(SRow) + ntp;
break;
case (HAS_NULL | HAS_NONE):
ntp = sizeof(SRow) + BIT1_SIZE(pTSchema->numOfCols - 1);
break;
case (HAS_VALUE | HAS_NONE):
case (HAS_VALUE | HAS_NULL):
ntp = sizeof(SRow) + BIT1_SIZE(pTSchema->numOfCols - 1) + ntp;
break;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
ntp = sizeof(SRow) + BIT2_SIZE(pTSchema->numOfCols - 1) + ntp;
break;
default:
ASSERT(0);
break;
}
if (maxIdx <= UINT8_MAX) {
nkv = sizeof(SRow) + sizeof(SKVIdx) + nIdx + nkv;
flag |= KV_FLG_LIT;
} else if (maxIdx <= UINT16_MAX) {
nkv = sizeof(SRow) + sizeof(SKVIdx) + (nIdx << 1) + nkv;
flag |= KV_FLG_MID;
} else { } else {
switch (flags) { nkv = sizeof(SRow) + sizeof(SKVIdx) + (nIdx << 2) + nkv;
case TSROW_HAS_VAL: flag |= KV_FLG_BIG;
nDataT = pTSchema->flen + ntv; }
break; int32_t nRow;
case TSROW_HAS_NULL | TSROW_HAS_NONE: if (nkv < ntp) {
nDataT = BIT1_SIZE(pTSchema->numOfCols - 1); nRow = nkv;
break; } else {
case TSROW_HAS_VAL | TSROW_HAS_NONE: nRow = ntp;
case TSROW_HAS_VAL | TSROW_HAS_NULL: flag &= ((uint8_t)0x0f);
nDataT = BIT1_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntv;
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
nDataT = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntv;
break;
default:
break;
ASSERT(0);
}
uint8_t tflags = 0;
if (maxIdx <= UINT8_MAX) {
nDataK = sizeof(STSKVRow) + sizeof(uint8_t) * nTag + nkv;
tflags |= TSROW_KV_SMALL;
} else if (maxIdx <= UINT16_MAX) {
nDataK = sizeof(STSKVRow) + sizeof(uint16_t) * nTag + nkv;
tflags |= TSROW_KV_MID;
} else {
nDataK = sizeof(STSKVRow) + sizeof(uint32_t) * nTag + nkv;
tflags |= TSROW_KV_BIG;
}
if (nDataT <= nDataK) {
nData = nDataT;
} else {
nData = nDataK;
flags |= tflags;
}
} }
// alloc // alloc --------------
if (pBuilder) { SRow *pRow = NULL;
// create from a builder code = tBufferReserve(pBuffer, nRow, (void **)&pRow);
if (nData == 0) { if (code) return code;
pBuilder->tsRow.nData = 0;
pBuilder->tsRow.pData = NULL;
} else {
if (pBuilder->szBuf < nData) {
uint8_t *p = taosMemoryRealloc(pBuilder->pBuf, nData);
if (p == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pBuilder->pBuf = p;
pBuilder->szBuf = nData;
}
pBuilder->tsRow.nData = nData; // build --------------
pBuilder->tsRow.pData = pBuilder->pBuf; pColVal = (SColVal *)taosArrayGet(aColVal, 0);
}
*ppRow = &pBuilder->tsRow; pRow->flag = flag;
} else { pRow->rsv = 0;
// create a new one pRow->sver = pTSchema->version;
*ppRow = (STSRow2 *)taosMemoryMalloc(sizeof(STSRow2)); pRow->len = nRow;
if (*ppRow == NULL) { memcpy(&pRow->ts, &pColVal->value.val, sizeof(TSKEY));
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; if (flag == HAS_NONE || flag == HAS_NULL) {
} goto _exit;
if (nData == 0) {
(*ppRow)->nData = 0;
(*ppRow)->pData = NULL;
} else {
(*ppRow)->nData = nData;
(*ppRow)->pData = taosMemoryMalloc(nData);
if ((*ppRow)->pData == NULL) {
taosMemoryFree(*ppRow);
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
} }
// build iColVal = 1;
(*ppRow)->flags = flags; pColVal = (iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
(*ppRow)->sver = pTSchema->version; iTColumn = 1;
pTColumn = pTSchema->columns + iTColumn;
pColVal = (SColVal *)taosArrayGet(pArray, 0); if (flag & 0xf0) { // KV
(*ppRow)->ts = pColVal->value.ts; SKVIdx *pIdx = (SKVIdx *)pRow->data;
int32_t iIdx = 0;
if ((*ppRow)->pData) { int32_t nv = 0;
STSKVRow *pTSKVRow = NULL; uint8_t *pv = NULL;
uint8_t *pidx = NULL; if (flag & KV_FLG_LIT) {
uint8_t *pkv = NULL; pv = pIdx->idx + nIdx;
uint8_t *pb = NULL; } else if (flag & KV_FLG_MID) {
uint8_t *pf = NULL; pv = pIdx->idx + (nIdx << 1);
uint8_t *ptv = NULL;
nkv = 0;
ntv = 0;
iColVal = 1;
if ((flags & 0xf0) == 0) {
switch (flags & 0xf) {
case TSROW_HAS_VAL:
pf = (*ppRow)->pData;
ptv = pf + pTSchema->flen;
break;
case TSROW_HAS_NULL | TSROW_HAS_NONE:
pb = (*ppRow)->pData;
break;
case TSROW_HAS_VAL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NULL:
pb = (*ppRow)->pData;
pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
ptv = pf + pTSchema->flen;
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
pb = (*ppRow)->pData;
pf = pb + BIT2_SIZE(pTSchema->numOfCols - 1);
ptv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
}
} else { } else {
pTSKVRow = (STSKVRow *)(*ppRow)->pData; pv = pIdx->idx + (nIdx << 2);
pTSKVRow->nCols = 0;
pidx = pTSKVRow->idx;
if (flags & TSROW_KV_SMALL) {
pkv = pidx + sizeof(uint8_t) * nTag;
} else if (flags & TSROW_KV_MID) {
pkv = pidx + sizeof(uint16_t) * nTag;
} else {
pkv = pidx + sizeof(uint32_t) * nTag;
}
} }
pIdx->nCol = nIdx;
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) { while (pTColumn) {
pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else {
pColVal = NULL;
}
if (pColVal) { if (pColVal) {
if (pColVal->cid == pTColumn->colId) { if (pColVal->cid == pTColumn->colId) {
iColVal++; if (COL_VAL_IS_VALUE(pColVal)) {
if (flag & KV_FLG_LIT) {
((uint8_t *)pIdx->idx)[iIdx] = (uint8_t)nv;
} else if (flag & KV_FLG_MID) {
((uint16_t *)pIdx->idx)[iIdx] = (uint16_t)nv;
} else {
((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv;
}
iIdx++;
if (COL_VAL_IS_NONE(pColVal)) { nv += tPutI16v(pv + nv, pTColumn->colId);
goto _set_none; if (IS_VAR_DATA_TYPE(pTColumn->type)) {
nv += tPutU32v(pv + nv, pColVal->value.nData);
memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData);
nv += pColVal->value.nData;
} else {
memcpy(pv + nv, &pColVal->value.val, pTColumn->bytes);
nv += pTColumn->bytes;
}
} else if (COL_VAL_IS_NULL(pColVal)) { } else if (COL_VAL_IS_NULL(pColVal)) {
goto _set_null; if (flag & KV_FLG_LIT) {
} else { ((uint8_t *)pIdx->idx)[iIdx] = (uint8_t)nv;
goto _set_value; } else if (flag & KV_FLG_MID) {
((uint16_t *)pIdx->idx)[iIdx] = (uint16_t)nv;
} else {
((uint32_t *)pIdx->idx)[iIdx] = (uint32_t)nv;
}
iIdx++;
nv += tPutI16v(pv + nv, -pTColumn->colId);
} }
} else if (pColVal->cid > pTColumn->colId) {
goto _set_none; pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
} else if (pColVal->cid > pTColumn->colId) { // NONE
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
} else { } else {
ASSERT(0); pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
} }
} else { } else { // NONE
goto _set_none; pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
} }
}
} else { // TUPLE
uint8_t *pb = NULL;
uint8_t *pf = NULL;
uint8_t *pv = NULL;
int32_t nv = 0;
_set_none: switch (flag) {
if ((flags & 0xf0) == 0) { case (HAS_NULL | HAS_NONE):
setBitMap(pb, 0, iColumn - 1, flags); pb = pRow->data;
if (flags & TSROW_HAS_VAL) { // set 0 break;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { case HAS_VALUE:
*(VarDataOffsetT *)(pf + pTColumn->offset) = 0; pf = pRow->data;
} else { pv = pf + pTSchema->flen;
tPutValue(pf + pTColumn->offset, &((SValue){0}), pTColumn->type); break;
} case (HAS_VALUE | HAS_NONE):
} case (HAS_VALUE | HAS_NULL):
} pb = pRow->data;
continue; pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
pv = pf + pTSchema->flen;
break;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
pb = pRow->data;
pf = pb + BIT2_SIZE(pTSchema->numOfCols - 1);
pv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
}
_set_null: // build impl
if ((flags & 0xf0) == 0) { while (pTColumn) {
setBitMap(pb, 1, iColumn - 1, flags); if (pColVal) {
if (flags & TSROW_HAS_VAL) { // set 0 if (pColVal->cid == pTColumn->colId) {
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (COL_VAL_IS_VALUE(pColVal)) { // VALUE
*(VarDataOffsetT *)(pf + pTColumn->offset) = 0; ROW_SET_BITMAP(pb, flag, iTColumn - 1, ROW_BIT_VALUE);
} else {
tPutValue(pf + pTColumn->offset, &((SValue){0}), pTColumn->type);
}
}
} else {
SET_IDX(pidx, pTSKVRow->nCols, nkv, flags);
pTSKVRow->nCols++;
nkv += tPutI16v(pkv + nkv, -pTColumn->colId);
}
continue;
_set_value: if (IS_VAR_DATA_TYPE(pTColumn->type)) {
if ((flags & 0xf0) == 0) { *(int32_t *)(pf + pTColumn->offset) = nv;
setBitMap(pb, 2, iColumn - 1, flags); nv += tPutU32v(pv + nv, pColVal->value.nData);
if (pColVal->value.nData) {
memcpy(pv + nv, pColVal->value.pData, pColVal->value.nData);
nv += pColVal->value.nData;
}
} else {
memcpy(pf + pTColumn->offset, &pColVal->value.val, TYPE_BYTES[pTColumn->type]);
}
} else if (COL_VAL_IS_NONE(pColVal)) { // NONE
ROW_SET_BITMAP(pb, flag, iTColumn - 1, ROW_BIT_NONE);
if (pf) memset(pf + pTColumn->offset, 0, TYPE_BYTES[pTColumn->type]);
} else { // NULL
ROW_SET_BITMAP(pb, flag, iTColumn - 1, ROW_BIT_NULL);
if (pf) memset(pf + pTColumn->offset, 0, TYPE_BYTES[pTColumn->type]);
}
if (IS_VAR_DATA_TYPE(pTColumn->type)) { pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
*(VarDataOffsetT *)(pf + pTColumn->offset) = ntv; pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
ntv += tPutValue(ptv + ntv, &pColVal->value, pTColumn->type); } else if (pColVal->cid > pTColumn->colId) { // NONE
ROW_SET_BITMAP(pb, flag, iTColumn - 1, ROW_BIT_NONE);
if (pf) memset(pf + pTColumn->offset, 0, TYPE_BYTES[pTColumn->type]);
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
} else { } else {
tPutValue(pf + pTColumn->offset, &pColVal->value, pTColumn->type); pColVal = (++iColVal < nColVal) ? (SColVal *)taosArrayGet(aColVal, iColVal) : NULL;
} }
} else { } else { // NONE
SET_IDX(pidx, pTSKVRow->nCols, nkv, flags); ROW_SET_BITMAP(pb, flag, iTColumn - 1, ROW_BIT_NONE);
pTSKVRow->nCols++; if (pf) memset(pf + pTColumn->offset, 0, TYPE_BYTES[pTColumn->type]);
nkv += tPutI16v(pkv + nkv, pColVal->cid); pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
nkv += tPutValue(pkv + nkv, &pColVal->value, pTColumn->type);
} }
continue;
} }
} }
#endif
_exit: _exit:
return code; return code;
} }
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow) { void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
int32_t code = 0;
int32_t rLen = 0;
TSROW_LEN(pRow, rLen);
(*ppRow) = (STSRow2 *)taosMemoryMalloc(rLen);
if (*ppRow == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
memcpy(*ppRow, pRow, rLen);
_exit:
return code;
}
void tTSRowFree(STSRow2 *pRow) {
if (pRow) {
taosMemoryFree(pRow);
}
}
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
uint8_t isTuple = ((pRow->flags & 0xf0) == 0) ? 1 : 0;
STColumn *pTColumn = &pTSchema->columns[iCol];
uint8_t flags = pRow->flags & (uint8_t)0xf;
SValue value;
ASSERT(iCol < pTSchema->numOfCols); ASSERT(iCol < pTSchema->numOfCols);
ASSERT(flags);
ASSERT(pRow->sver == pTSchema->version); ASSERT(pRow->sver == pTSchema->version);
STColumn *pTColumn = pTSchema->columns + iCol;
if (iCol == 0) { if (iCol == 0) {
value.ts = pRow->ts; pColVal->cid = pTColumn->colId;
goto _return_value; pColVal->type = pTColumn->type;
pColVal->flag = CV_FLAG_VALUE;
memcpy(&pColVal->value.val, &pRow->ts, sizeof(TSKEY));
return;
} }
if (flags == TSROW_HAS_NONE) { if (pRow->flag == HAS_NONE) {
goto _return_none; *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
} else if (flags == TSROW_HAS_NULL) { return;
goto _return_null;
} }
ASSERT(pRow->nData && pRow->pData); if (pRow->flag == HAS_NULL) {
*pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
return;
}
if (isTuple) { if (pRow->flag & 0xf0) { // KV Row
uint8_t *pb = pRow->pData; SKVIdx *pIdx = (SKVIdx *)pRow->data;
uint8_t *pf = NULL;
uint8_t *pv = NULL; uint8_t *pv = NULL;
uint8_t *p; if (pRow->flag & KV_FLG_LIT) {
uint8_t b; pv = pIdx->idx + pIdx->nCol;
} else if (pRow->flag & KV_FLG_MID) {
// bit pv = pIdx->idx + (pIdx->nCol << 1);
switch (flags) {
case TSROW_HAS_VAL:
pf = pb;
break;
case TSROW_HAS_NULL | TSROW_HAS_NONE:
b = GET_BIT1(pb, iCol - 1);
if (b == 0) {
goto _return_none;
} else {
goto _return_null;
}
case TSROW_HAS_VAL | TSROW_HAS_NONE:
b = GET_BIT1(pb, iCol - 1);
if (b == 0) {
goto _return_none;
} else {
pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
break;
}
case TSROW_HAS_VAL | TSROW_HAS_NULL:
b = GET_BIT1(pb, iCol - 1);
if (b == 0) {
goto _return_null;
} else {
pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
break;
}
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
b = GET_BIT2(pb, iCol - 1);
if (b == 0) {
goto _return_none;
} else if (b == 1) {
goto _return_null;
} else {
pf = pb + BIT2_SIZE(pTSchema->numOfCols - 1);
break;
}
default:
ASSERT(0);
}
ASSERT(pf);
p = pf + pTColumn->offset;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pv = pf + pTSchema->flen;
p = pv + *(VarDataOffsetT *)p;
}
tGetValue(p, &value, pTColumn->type);
goto _return_value;
} else {
STSKVRow *pRowK = (STSKVRow *)pRow->pData;
int16_t lidx = 0;
int16_t ridx = pRowK->nCols - 1;
uint8_t *p;
int16_t midx;
uint32_t n;
int16_t cid;
ASSERT(pRowK->nCols > 0);
if (pRow->flags & TSROW_KV_SMALL) {
p = pRow->pData + sizeof(STSKVRow) + sizeof(uint8_t) * pRowK->nCols;
} else if (pRow->flags & TSROW_KV_MID) {
p = pRow->pData + sizeof(STSKVRow) + sizeof(uint16_t) * pRowK->nCols;
} else if (pRow->flags & TSROW_KV_BIG) {
p = pRow->pData + sizeof(STSKVRow) + sizeof(uint32_t) * pRowK->nCols;
} else { } else {
ASSERT(0); pv = pIdx->idx + (pIdx->nCol << 2);
} }
while (lidx <= ridx) {
midx = (lidx + ridx) / 2;
if (pRow->flags & TSROW_KV_SMALL) { int16_t lidx = 0;
n = ((uint8_t *)pRowK->idx)[midx]; int16_t ridx = pIdx->nCol - 1;
} else if (pRow->flags & TSROW_KV_MID) { while (lidx <= ridx) {
n = ((uint16_t *)pRowK->idx)[midx]; int16_t mid = (lidx + ridx) >> 1;
uint8_t *pData = NULL;
if (pRow->flag & KV_FLG_LIT) {
pData = pv + ((uint8_t *)pIdx->idx)[mid];
} else if (pRow->flag & KV_FLG_MID) {
pData = pv + ((uint16_t *)pIdx->idx)[mid];
} else { } else {
n = ((uint32_t *)pRowK->idx)[midx]; pData = pv + ((uint32_t *)pIdx->idx)[mid];
} }
n += tGetI16v(p + n, &cid); int16_t cid;
pData += tGetI16v(pData, &cid);
if (TABS(cid) == pTColumn->colId) { if (TABS(cid) == pTColumn->colId) {
if (cid < 0) { if (cid < 0) {
goto _return_null; *pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
} else { } else {
n += tGetValue(p + n, &value, pTColumn->type); pColVal->cid = pTColumn->colId;
goto _return_value; pColVal->type = pTColumn->type;
} pColVal->flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pData += tGetU32v(pData, &pColVal->value.nData);
if (pColVal->value.nData > 0) {
pColVal->value.pData = pData;
} else {
pColVal->value.pData = NULL;
}
} else {
memcpy(&pColVal->value.val, pData, pTColumn->bytes);
}
}
return; return;
} else if (TABS(cid) > pTColumn->colId) { } else if (TABS(cid) < pTColumn->colId) {
ridx = midx - 1; lidx = mid + 1;
} else { } else {
lidx = midx + 1; ridx = mid - 1;
} }
} }
// not found, return NONE *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
goto _return_none; } else { // Tuple Row
} uint8_t *pf = NULL;
uint8_t *pv = NULL;
uint8_t bv = ROW_BIT_VALUE;
_return_none: switch (pRow->flag) {
*pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type); case HAS_VALUE:
return; pf = pRow->data;
pv = pf + pTSchema->flen;
break;
case (HAS_NULL | HAS_NONE):
bv = GET_BIT1(pRow->data, iCol - 1);
break;
case (HAS_VALUE | HAS_NONE):
bv = GET_BIT1(pRow->data, iCol - 1);
if (bv) bv++;
pf = pRow->data + BIT1_SIZE(pTSchema->numOfCols - 1);
pv = pf + pTSchema->flen;
break;
case (HAS_VALUE | HAS_NULL):
bv = GET_BIT1(pRow->data, iCol - 1);
bv++;
pf = pRow->data + BIT1_SIZE(pTSchema->numOfCols - 1);
pv = pf + pTSchema->flen;
break;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
bv = GET_BIT2(pRow->data, iCol - 1);
pf = pRow->data + BIT2_SIZE(pTSchema->numOfCols - 1);
pv = pf + pTSchema->flen;
break;
default:
break;
}
_return_null: if (bv == ROW_BIT_NONE) {
*pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type); *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
return; return;
} else if (bv == ROW_BIT_NULL) {
*pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
return;
}
_return_value: pColVal->cid = pTColumn->colId;
*pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value); pColVal->type = pTColumn->type;
return; pColVal->flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
uint8_t *pData = pv + *(int32_t *)(pf + pTColumn->offset);
pData += tGetU32v(pData, &pColVal->value.nData);
if (pColVal->value.nData) {
pColVal->value.pData = pData;
} else {
pColVal->value.pData = NULL;
}
} else {
memcpy(&pColVal->value.val, pv + pTColumn->offset, pTColumn->bytes);
}
}
} }
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) { // SRowIter ========================================
int32_t code = 0; struct SRowIter {
SRow *pRow;
STSchema *pTSchema;
int32_t iTColumn;
union {
struct { // kv
int32_t iCol;
SKVIdx *pIdx;
};
struct { // tuple
uint8_t *pb;
uint8_t *pf;
};
uint8_t *pv;
};
SColVal cv; SColVal cv;
};
(*ppArray) = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal)); int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter) {
if (*ppArray == NULL) { ASSERT(pRow->sver == pTSchema->version);
code = TSDB_CODE_OUT_OF_MEMORY;
int32_t code = 0;
SRowIter *pIter = taosMemoryCalloc(1, sizeof(*pIter));
if (pIter == NULL) {
code = TSDB_CODE_TDB_OUT_OF_MEMORY;
goto _exit; goto _exit;
} }
for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; iColumn++) { pIter->pRow = pRow;
tTSRowGet(pRow, pTSchema, iColumn, &cv); pIter->pTSchema = pTSchema;
taosArrayPush(*ppArray, &cv); pIter->iTColumn = 0;
}
if (pRow->flag == HAS_NONE || pRow->flag == HAS_NULL) goto _exit;
if (pRow->flag & 0xf0) {
pIter->iCol = 0;
pIter->pIdx = (SKVIdx *)pRow->data;
if (pRow->flag & KV_FLG_LIT) {
pIter->pv = pIter->pIdx->idx + pIter->pIdx->nCol;
} else if (pRow->flag & KV_FLG_MID) {
pIter->pv = pIter->pIdx->idx + (pIter->pIdx->nCol << 1);
} else {
pIter->pv = pIter->pIdx->idx + (pIter->pIdx->nCol << 2);
}
} else {
switch (pRow->flag) {
case (HAS_NULL | HAS_NONE):
pIter->pb = pRow->data;
break;
case HAS_VALUE:
pIter->pf = pRow->data;
pIter->pv = pIter->pf + pTSchema->flen;
break;
case (HAS_VALUE | HAS_NONE):
case (HAS_VALUE | HAS_NULL):
pIter->pb = pRow->data;
pIter->pf = pRow->data + BIT1_SIZE(pTSchema->numOfCols - 1);
pIter->pv = pIter->pf + pTSchema->flen;
break;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
pIter->pb = pRow->data;
pIter->pf = pRow->data + BIT2_SIZE(pTSchema->numOfCols - 1);
pIter->pv = pIter->pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
}
}
_exit: _exit:
if (code) {
*ppIter = NULL;
if (pIter) taosMemoryFree(pIter);
} else {
*ppIter = pIter;
}
return code; return code;
} }
#endif
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
int32_t n = 0;
TSROW_LEN(pRow, n); void tRowIterClose(SRowIter **ppIter) {
if (p) { SRowIter *pIter = *ppIter;
memcpy(p, pRow, n); if (pIter) {
taosMemoryFree(pIter);
} }
*ppIter = NULL;
return n;
} }
int32_t tGetTSRow(uint8_t *p, STSRow2 **ppRow) { SColVal *tRowIterNext(SRowIter *pIter) {
int32_t n = 0; if (pIter->iTColumn >= pIter->pTSchema->numOfCols) {
return NULL;
}
*ppRow = (STSRow2 *)p; STColumn *pTColumn = pIter->pTSchema->columns + pIter->iTColumn;
TSROW_LEN(*ppRow, n);
return n; // timestamp
} if (0 == pIter->iTColumn) {
pIter->cv.cid = pTColumn->colId;
pIter->cv.type = pTColumn->type;
pIter->cv.flag = CV_FLAG_VALUE;
memcpy(&pIter->cv.value.val, &pIter->pRow->ts, sizeof(TSKEY));
goto _exit;
}
// STSchema ======================================== if (pIter->pRow->flag == HAS_NONE) {
int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t ncols, STSchema **ppTSchema) { pIter->cv = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
*ppTSchema = (STSchema *)taosMemoryMalloc(sizeof(STSchema) + sizeof(STColumn) * ncols); goto _exit;
if (*ppTSchema == NULL) { }
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; if (pIter->pRow->flag == HAS_NULL) {
pIter->cv = COL_VAL_NULL(pTColumn->type, pTColumn->colId);
goto _exit;
} }
(*ppTSchema)->numOfCols = ncols; if (pIter->pRow->flag & 0xf0) { // KV
(*ppTSchema)->version = sver; if (pIter->iCol < pIter->pIdx->nCol) {
(*ppTSchema)->flen = 0; uint8_t *pData;
(*ppTSchema)->vlen = 0;
(*ppTSchema)->tlen = 0;
for (int32_t iCol = 0; iCol < ncols; iCol++) { if (pIter->pRow->flag & KV_FLG_LIT) {
SSchema *pColumn = &pSchema[iCol]; pData = pIter->pv + ((uint8_t *)pIter->pIdx->idx)[pIter->iCol];
STColumn *pTColumn = &((*ppTSchema)->columns[iCol]); } else if (pIter->pRow->flag & KV_FLG_MID) {
pData = pIter->pv + ((uint16_t *)pIter->pIdx->idx)[pIter->iCol];
} else {
pData = pIter->pv + ((uint32_t *)pIter->pIdx->idx)[pIter->iCol];
}
int16_t cid;
pData += tGetI16v(pData, &cid);
pTColumn->colId = pColumn->colId; if (TABS(cid) == pTColumn->colId) {
pTColumn->type = pColumn->type; if (cid < 0) {
pTColumn->flags = pColumn->flags; pIter->cv = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
pTColumn->bytes = pColumn->bytes; } else {
pTColumn->offset = (*ppTSchema)->flen; pIter->cv.cid = pTColumn->colId;
pIter->cv.type = pTColumn->type;
pIter->cv.flag = CV_FLAG_VALUE;
// skip first column if (IS_VAR_DATA_TYPE(pTColumn->type)) {
if (iCol) { pData += tGetU32v(pData, &pIter->cv.value.nData);
(*ppTSchema)->flen += TYPE_BYTES[pColumn->type]; if (pIter->cv.value.nData > 0) {
if (IS_VAR_DATA_TYPE(pColumn->type)) { pIter->cv.value.pData = pData;
(*ppTSchema)->vlen += (pColumn->bytes + 5); } else {
pIter->cv.value.pData = NULL;
}
} else {
memcpy(&pIter->cv.value.val, pData, pTColumn->bytes);
}
}
pIter->iCol++;
goto _exit;
} else if (TABS(cid) > pTColumn->colId) {
pIter->cv = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
goto _exit;
} else {
ASSERT(0);
} }
} else {
pIter->cv = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
goto _exit;
} }
} else { // Tuple
uint8_t bv = ROW_BIT_VALUE;
if (pIter->pb) {
switch (pIter->pRow->flag) {
case (HAS_NULL | HAS_NONE):
bv = GET_BIT1(pIter->pb, pIter->iTColumn - 1);
break;
case (HAS_VALUE | HAS_NONE):
bv = GET_BIT1(pIter->pb, pIter->iTColumn - 1);
if (bv) bv++;
break;
case (HAS_VALUE | HAS_NULL):
bv = GET_BIT1(pIter->pb, pIter->iTColumn - 1) + 1;
break;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
bv = GET_BIT2(pIter->pb, pIter->iTColumn - 1);
break;
default:
ASSERT(0);
break;
}
if (bv == ROW_BIT_NONE) {
pIter->cv = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
goto _exit;
} else if (bv == ROW_BIT_NULL) {
pIter->cv = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
goto _exit;
}
}
pIter->cv.cid = pTColumn->colId;
pIter->cv.type = pTColumn->type;
pIter->cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
uint8_t *pData = pIter->pv + *(int32_t *)(pIter->pf + pTColumn->offset);
pData += tGetU32v(pData, &pIter->cv.value.nData);
if (pIter->cv.value.nData > 0) {
pIter->cv.value.pData = pData;
} else {
pIter->cv.value.pData = NULL;
}
} else {
memcpy(&pIter->cv.value.val, pIter->pv + pTColumn->offset, pTColumn->bytes);
}
goto _exit;
} }
return 0; _exit:
pIter->iTColumn++;
return &pIter->cv;
} }
void tTSchemaDestroy(STSchema *pTSchema) { // STSchema ========================================
if (pTSchema) taosMemoryFree(pTSchema);
}
// STag ======================================== // STag ========================================
static int tTagValCmprFn(const void *p1, const void *p2) { static int tTagValCmprFn(const void *p1, const void *p2) {
...@@ -1042,7 +1108,6 @@ void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) { ...@@ -1042,7 +1108,6 @@ void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
pBuilder->nCols = 0; pBuilder->nCols = 0;
pBuilder->tlen = 0; pBuilder->tlen = 0;
pBuilder->flen = 0; pBuilder->flen = 0;
pBuilder->vlen = 0;
pBuilder->version = version; pBuilder->version = version;
} }
...@@ -1061,24 +1126,21 @@ int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, c ...@@ -1061,24 +1126,21 @@ int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, c
pCol->colId = colId; pCol->colId = colId;
pCol->flags = flags; pCol->flags = flags;
if (pBuilder->nCols == 0) { if (pBuilder->nCols == 0) {
pCol->offset = 0; pCol->offset = -1;
} else { } else {
STColumn *pTCol = &(pBuilder->columns[pBuilder->nCols - 1]); pCol->offset = pBuilder->flen;
pCol->offset = pTCol->offset + TYPE_BYTES[pTCol->type]; pBuilder->flen += TYPE_BYTES[type];
} }
if (IS_VAR_DATA_TYPE(type)) { if (IS_VAR_DATA_TYPE(type)) {
pCol->bytes = bytes; pCol->bytes = bytes;
pBuilder->tlen += (TYPE_BYTES[type] + bytes); pBuilder->tlen += (TYPE_BYTES[type] + bytes);
pBuilder->vlen += bytes - sizeof(VarDataLenT);
} else { } else {
pCol->bytes = TYPE_BYTES[type]; pCol->bytes = TYPE_BYTES[type];
pBuilder->tlen += TYPE_BYTES[type]; pBuilder->tlen += TYPE_BYTES[type];
pBuilder->vlen += TYPE_BYTES[type];
} }
pBuilder->nCols++; pBuilder->nCols++;
pBuilder->flen += TYPE_BYTES[type];
ASSERT(pCol->offset < pBuilder->flen); ASSERT(pCol->offset < pBuilder->flen);
...@@ -1097,7 +1159,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) { ...@@ -1097,7 +1159,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
pSchema->numOfCols = pBuilder->nCols; pSchema->numOfCols = pBuilder->nCols;
pSchema->tlen = pBuilder->tlen; pSchema->tlen = pBuilder->tlen;
pSchema->flen = pBuilder->flen; pSchema->flen = pBuilder->flen;
pSchema->vlen = pBuilder->vlen;
#ifdef TD_SUPPORT_BITMAP #ifdef TD_SUPPORT_BITMAP
pSchema->tlen += (int)TD_BITMAP_BYTES(pSchema->numOfCols); pSchema->tlen += (int)TD_BITMAP_BYTES(pSchema->numOfCols);
...@@ -1110,6 +1171,43 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) { ...@@ -1110,6 +1171,43 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
#endif #endif
STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version) {
STSchema *pTSchema = taosMemoryCalloc(1, sizeof(STSchema) + sizeof(STColumn) * numOfCols);
if (pTSchema == NULL) return NULL;
pTSchema->numOfCols = numOfCols;
pTSchema->version = version;
// timestamp column
ASSERT(aSchema[0].type == TSDB_DATA_TYPE_TIMESTAMP);
ASSERT(aSchema[0].colId == PRIMARYKEY_TIMESTAMP_COL_ID);
pTSchema->columns[0].colId = aSchema[0].colId;
pTSchema->columns[0].type = aSchema[0].type;
pTSchema->columns[0].flags = aSchema[0].flags;
pTSchema->columns[0].bytes = aSchema[0].bytes;
pTSchema->columns[0].offset = -1;
// other columns
for (int32_t iCol = 1; iCol < numOfCols; iCol++) {
SSchema *pSchema = &aSchema[iCol];
STColumn *pTColumn = &pTSchema->columns[iCol];
pTColumn->colId = pSchema->colId;
pTColumn->type = pSchema->type;
pTColumn->flags = pSchema->flags;
pTColumn->bytes = pSchema->bytes;
pTColumn->offset = pTSchema->flen;
pTSchema->flen += TYPE_BYTES[pTColumn->type];
}
return pTSchema;
}
void tDestroyTSchema(STSchema *pTSchema) {
if (pTSchema) taosMemoryFree(pTSchema);
}
// SColData ======================================== // SColData ========================================
void tColDataDestroy(void *ph) { void tColDataDestroy(void *ph) {
SColData *pColData = (SColData *)ph; SColData *pColData = (SColData *)ph;
...@@ -1496,7 +1594,7 @@ static FORCE_INLINE void tColDataGetValue4(SColData *pColData, int32_t iVal, SCo ...@@ -1496,7 +1594,7 @@ static FORCE_INLINE void tColDataGetValue4(SColData *pColData, int32_t iVal, SCo
} }
value.pData = pColData->pData + pColData->aOffset[iVal]; value.pData = pColData->pData + pColData->aOffset[iVal];
} else { } else {
tGetValue(pColData->pData + tDataTypes[pColData->type].bytes * iVal, &value, pColData->type); memcpy(&value.val, pColData->pData + tDataTypes[pColData->type].bytes * iVal, tDataTypes[pColData->type].bytes);
} }
*pColVal = COL_VAL_VALUE(pColData->cid, pColData->type, value); *pColVal = COL_VAL_VALUE(pColData->cid, pColData->type, value);
} }
......
source/common/src/trow.c
浏览文件 @ f58396ea
...@@ -192,7 +192,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl ...@@ -192,7 +192,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl
return true; return true;
} }
void *pBitmap = tdGetBitmapAddrTp(pRow, flen); void *pBitmap = tdGetBitmapAddrTp(pRow, flen);
tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset - sizeof(TSKEY), colIdx); tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset, colIdx);
return true; return true;
} }
...@@ -217,7 +217,7 @@ bool tdSTSRowIterFetch(STSRowIter *pIter, col_id_t colId, col_type_t colType, SC ...@@ -217,7 +217,7 @@ bool tdSTSRowIterFetch(STSRowIter *pIter, col_id_t colId, col_type_t colType, SC
return false; return false;
} }
} }
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal); tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
++pIter->colIdx; ++pIter->colIdx;
} else if (TD_IS_KV_ROW(pIter->pRow)) { } else if (TD_IS_KV_ROW(pIter->pRow)) {
return tdSTSRowIterGetKvVal(pIter, colId, &pIter->kvIdx, pVal); return tdSTSRowIterGetKvVal(pIter, colId, &pIter->kvIdx, pVal);
...@@ -244,7 +244,7 @@ bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) { ...@@ -244,7 +244,7 @@ bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) {
} }
if (TD_IS_TP_ROW(pIter->pRow)) { if (TD_IS_TP_ROW(pIter->pRow)) {
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal); tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
} else if (TD_IS_KV_ROW(pIter->pRow)) { } else if (TD_IS_KV_ROW(pIter->pRow)) {
tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal); tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
} else { } else {
...@@ -469,7 +469,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell ...@@ -469,7 +469,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
#ifdef TD_SUPPORT_BITMAP #ifdef TD_SUPPORT_BITMAP
colIdx = POINTER_DISTANCE(pCol, pSchema->columns) / sizeof(STColumn); colIdx = POINTER_DISTANCE(pCol, pSchema->columns) / sizeof(STColumn);
#endif #endif
tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset - sizeof(TSKEY), colIdx - 1); tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset, colIdx - 1);
} else if (TD_IS_KV_ROW(pRow)) { } else if (TD_IS_KV_ROW(pRow)) {
SKvRowIdx *pIdx = (SKvRowIdx *)taosbsearch(&colId, TD_ROW_COL_IDX(pRow), tdRowGetNCols(pRow), sizeof(SKvRowIdx), SKvRowIdx *pIdx = (SKvRowIdx *)taosbsearch(&colId, TD_ROW_COL_IDX(pRow), tdRowGetNCols(pRow), sizeof(SKvRowIdx),
compareKvRowColId, TD_EQ); compareKvRowColId, TD_EQ);
...@@ -757,11 +757,10 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo ...@@ -757,11 +757,10 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val, bool isCopyVarData, int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val, bool isCopyVarData,
int8_t colType, int16_t colIdx, int32_t offset) { int8_t colType, int16_t colIdx, int32_t offset) {
if ((offset < (int32_t)sizeof(TSKEY)) || (colIdx < 1)) { if (colIdx < 1) {
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
return terrno; return terrno;
} }
offset -= sizeof(TSKEY);
--colIdx; --colIdx;
#ifdef TD_SUPPORT_BITMAP #ifdef TD_SUPPORT_BITMAP
...@@ -853,7 +852,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) { ...@@ -853,7 +852,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBitmaps); memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBitmaps);
#endif #endif
// the primary TS key is stored separatedly // the primary TS key is stored separatedly
len = TD_ROW_HEAD_LEN + pBuilder->flen - sizeof(TSKEY) + pBuilder->nBitmaps; len = TD_ROW_HEAD_LEN + pBuilder->flen + pBuilder->nBitmaps;
TD_ROW_SET_LEN(pBuilder->pBuf, len); TD_ROW_SET_LEN(pBuilder->pBuf, len);
TD_ROW_SET_SVER(pBuilder->pBuf, pBuilder->sver); TD_ROW_SET_SVER(pBuilder->pBuf, pBuilder->sver);
break; break;
......
source/common/src/ttypes.c
浏览文件 @ f58396ea
...@@ -61,7 +61,7 @@ tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = { ...@@ -61,7 +61,7 @@ tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = {
static float floatMin = -FLT_MAX, floatMax = FLT_MAX; static float floatMin = -FLT_MAX, floatMax = FLT_MAX;
static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX; static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX;
FORCE_INLINE void *getDataMin(int32_t type, void* value) { FORCE_INLINE void *getDataMin(int32_t type, void *value) {
switch (type) { switch (type) {
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
*(float *)value = floatMin; *(float *)value = floatMin;
...@@ -77,7 +77,7 @@ FORCE_INLINE void *getDataMin(int32_t type, void* value) { ...@@ -77,7 +77,7 @@ FORCE_INLINE void *getDataMin(int32_t type, void* value) {
return value; return value;
} }
FORCE_INLINE void *getDataMax(int32_t type, void* value) { FORCE_INLINE void *getDataMax(int32_t type, void *value) {
switch (type) { switch (type) {
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
*(float *)value = floatMax; *(float *)value = floatMax;
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ f58396ea
...@@ -56,7 +56,7 @@ typedef struct SDataFWriter SDataFWriter; ...@@ -56,7 +56,7 @@ typedef struct SDataFWriter SDataFWriter;
typedef struct SDataFReader SDataFReader; typedef struct SDataFReader SDataFReader;
typedef struct SDelFWriter SDelFWriter; typedef struct SDelFWriter SDelFWriter;
typedef struct SDelFReader SDelFReader; typedef struct SDelFReader SDelFReader;
typedef struct SRowIter SRowIter; typedef struct STSDBRowIter STSDBRowIter;
typedef struct STsdbFS STsdbFS; typedef struct STsdbFS STsdbFS;
typedef struct SRowMerger SRowMerger; typedef struct SRowMerger SRowMerger;
typedef struct STsdbReadSnap STsdbReadSnap; typedef struct STsdbReadSnap STsdbReadSnap;
...@@ -111,9 +111,9 @@ static FORCE_INLINE int64_t tsdbLogicToFileSize(int64_t lSize, int32_t szPage) { ...@@ -111,9 +111,9 @@ static FORCE_INLINE int64_t tsdbLogicToFileSize(int64_t lSize, int32_t szPage) {
void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal); void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
// int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow); // int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
int32_t tsdbRowCmprFn(const void *p1, const void *p2); int32_t tsdbRowCmprFn(const void *p1, const void *p2);
// SRowIter // STSDBRowIter
void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema); void tsdbRowIterInit(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema);
SColVal *tRowIterNext(SRowIter *pIter); SColVal *tsdbRowIterNext(STSDBRowIter *pIter);
// SRowMerger // SRowMerger
int32_t tRowMergerInit2(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema); int32_t tRowMergerInit2(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema);
int32_t tRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema); int32_t tRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
...@@ -562,7 +562,7 @@ struct SDFileSet { ...@@ -562,7 +562,7 @@ struct SDFileSet {
SSttFile *aSttF[TSDB_MAX_STT_TRIGGER]; SSttFile *aSttF[TSDB_MAX_STT_TRIGGER];
}; };
struct SRowIter { struct STSDBRowIter {
TSDBROW *pRow; TSDBROW *pRow;
STSchema *pTSchema; STSchema *pTSchema;
SColVal colVal; SColVal colVal;
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ f58396ea
...@@ -341,7 +341,7 @@ int32_t tsdbUpdateTableSchema(SMeta *pMeta, int64_t suid, int64_t uid, SSkmInfo ...@@ -341,7 +341,7 @@ int32_t tsdbUpdateTableSchema(SMeta *pMeta, int64_t suid, int64_t uid, SSkmInfo
pSkmInfo->suid = suid; pSkmInfo->suid = suid;
pSkmInfo->uid = uid; pSkmInfo->uid = uid;
tTSchemaDestroy(pSkmInfo->pTSchema); tDestroyTSchema(pSkmInfo->pTSchema);
code = metaGetTbTSchemaEx(pMeta, suid, uid, -1, &pSkmInfo->pTSchema); code = metaGetTbTSchemaEx(pMeta, suid, uid, -1, &pSkmInfo->pTSchema);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
...@@ -365,7 +365,7 @@ static int32_t tsdbCommitterUpdateRowSchema(SCommitter *pCommitter, int64_t suid ...@@ -365,7 +365,7 @@ static int32_t tsdbCommitterUpdateRowSchema(SCommitter *pCommitter, int64_t suid
pCommitter->skmRow.suid = suid; pCommitter->skmRow.suid = suid;
pCommitter->skmRow.uid = uid; pCommitter->skmRow.uid = uid;
tTSchemaDestroy(pCommitter->skmRow.pTSchema); tDestroyTSchema(pCommitter->skmRow.pTSchema);
code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmRow.pTSchema); code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmRow.pTSchema);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
...@@ -498,7 +498,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { ...@@ -498,7 +498,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
#if 0 #if 0
ASSERT(pCommitter->minKey <= pCommitter->nextKey && pCommitter->maxKey >= pCommitter->nextKey); ASSERT(pCommitter->minKey <= pCommitter->nextKey && pCommitter->maxKey >= pCommitter->nextKey);
#endif #endif
pCommitter->nextKey = TSKEY_MAX; pCommitter->nextKey = TSKEY_MAX;
// Reader // Reader
...@@ -623,7 +623,8 @@ int32_t tsdbWriteDataBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SMapDa ...@@ -623,7 +623,8 @@ int32_t tsdbWriteDataBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SMapDa
_exit: _exit:
if (code) { if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code)); tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
} }
return code; return code;
} }
...@@ -666,7 +667,8 @@ int32_t tsdbWriteSttBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SArray ...@@ -666,7 +667,8 @@ int32_t tsdbWriteSttBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SArray
_exit: _exit:
if (code) { if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code)); tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
} }
return code; return code;
} }
...@@ -706,7 +708,8 @@ static int32_t tsdbCommitSttBlk(SDataFWriter *pWriter, SDiskDataBuilder *pBuilde ...@@ -706,7 +708,8 @@ static int32_t tsdbCommitSttBlk(SDataFWriter *pWriter, SDiskDataBuilder *pBuilde
_exit: _exit:
if (code) { if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code)); tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
} }
return code; return code;
} }
...@@ -919,8 +922,8 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) { ...@@ -919,8 +922,8 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) {
#else #else
tBlockDataDestroy(&pCommitter->dWriter.bDatal, 1); tBlockDataDestroy(&pCommitter->dWriter.bDatal, 1);
#endif #endif
tTSchemaDestroy(pCommitter->skmTable.pTSchema); tDestroyTSchema(pCommitter->skmTable.pTSchema);
tTSchemaDestroy(pCommitter->skmRow.pTSchema); tDestroyTSchema(pCommitter->skmRow.pTSchema);
} }
static int32_t tsdbCommitData(SCommitter *pCommitter) { static int32_t tsdbCommitData(SCommitter *pCommitter) {
......
source/dnode/vnode/src/tsdb/tsdbDiskData.c
浏览文件 @ f58396ea
...@@ -595,21 +595,21 @@ int32_t tDiskDataAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTS ...@@ -595,21 +595,21 @@ int32_t tDiskDataAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTS
if (pBuilder->bi.minKey > kRow.ts) pBuilder->bi.minKey = kRow.ts; if (pBuilder->bi.minKey > kRow.ts) pBuilder->bi.minKey = kRow.ts;
if (pBuilder->bi.maxKey < kRow.ts) pBuilder->bi.maxKey = kRow.ts; if (pBuilder->bi.maxKey < kRow.ts) pBuilder->bi.maxKey = kRow.ts;
SRowIter iter = {0}; STSDBRowIter iter = {0};
tRowIterInit(&iter, pRow, pTSchema); tsdbRowIterInit(&iter, pRow, pTSchema);
SColVal *pColVal = tRowIterNext(&iter); SColVal *pColVal = tsdbRowIterNext(&iter);
for (int32_t iBuilder = 0; iBuilder < pBuilder->nBuilder; iBuilder++) { for (int32_t iBuilder = 0; iBuilder < pBuilder->nBuilder; iBuilder++) {
SDiskColBuilder *pDCBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iBuilder); SDiskColBuilder *pDCBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iBuilder);
while (pColVal && pColVal->cid < pDCBuilder->cid) { while (pColVal && pColVal->cid < pDCBuilder->cid) {
pColVal = tRowIterNext(&iter); pColVal = tsdbRowIterNext(&iter);
} }
if (pColVal && pColVal->cid == pDCBuilder->cid) { if (pColVal && pColVal->cid == pDCBuilder->cid) {
code = tDiskColAddVal(pDCBuilder, pColVal); code = tDiskColAddVal(pDCBuilder, pColVal);
if (code) return code; if (code) return code;
pColVal = tRowIterNext(&iter); pColVal = tsdbRowIterNext(&iter);
} else { } else {
code = tDiskColAddVal(pDCBuilder, &COL_VAL_NONE(pDCBuilder->cid, pDCBuilder->type)); code = tDiskColAddVal(pDCBuilder, &COL_VAL_NONE(pDCBuilder->cid, pDCBuilder->type));
if (code) return code; if (code) return code;
......
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
浏览文件 @ f58396ea
...@@ -555,7 +555,7 @@ int32_t tsdbSnapReaderClose(STsdbSnapReader** ppReader) { ...@@ -555,7 +555,7 @@ int32_t tsdbSnapReaderClose(STsdbSnapReader** ppReader) {
} }
tBlockDataDestroy(&pReader->bData, 1); tBlockDataDestroy(&pReader->bData, 1);
tTSchemaDestroy(pReader->skmTable.pTSchema); tDestroyTSchema(pReader->skmTable.pTSchema);
// del // del
if (pReader->pDelFReader) tsdbDelFReaderClose(&pReader->pDelFReader); if (pReader->pDelFReader) tsdbDelFReaderClose(&pReader->pDelFReader);
...@@ -1416,7 +1416,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) { ...@@ -1416,7 +1416,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
taosArrayDestroy(pWriter->dReader.aBlockIdx); taosArrayDestroy(pWriter->dReader.aBlockIdx);
tBlockDataDestroy(&pWriter->bData, 1); tBlockDataDestroy(&pWriter->bData, 1);
tTSchemaDestroy(pWriter->skmTable.pTSchema); tDestroyTSchema(pWriter->skmTable.pTSchema);
for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t*); iBuf++) { for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t*); iBuf++) {
tFree(pWriter->aBuf[iBuf]); tFree(pWriter->aBuf[iBuf]);
......
source/dnode/vnode/src/tsdb/tsdbUtil.c
浏览文件 @ f58396ea
...@@ -579,8 +579,8 @@ int32_t tsdbRowCmprFn(const void *p1, const void *p2) { ...@@ -579,8 +579,8 @@ int32_t tsdbRowCmprFn(const void *p1, const void *p2) {
return tsdbKeyCmprFn(&TSDBROW_KEY((TSDBROW *)p1), &TSDBROW_KEY((TSDBROW *)p2)); return tsdbKeyCmprFn(&TSDBROW_KEY((TSDBROW *)p1), &TSDBROW_KEY((TSDBROW *)p2));
} }
// SRowIter ====================================================== // STSDBRowIter ======================================================
void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) { void tsdbRowIterInit(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
pIter->pRow = pRow; pIter->pRow = pRow;
if (pRow->type == 0) { if (pRow->type == 0) {
ASSERT(pTSchema); ASSERT(pTSchema);
...@@ -594,7 +594,7 @@ void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) { ...@@ -594,7 +594,7 @@ void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
} }
} }
SColVal *tRowIterNext(SRowIter *pIter) { SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
if (pIter->pRow->type == 0) { if (pIter->pRow->type == 0) {
if (pIter->i < pIter->pTSchema->numOfCols) { if (pIter->i < pIter->pTSchema->numOfCols) {
tTSRowGetVal(pIter->pRow->pTSRow, pIter->pTSchema, pIter->i, &pIter->colVal); tTSRowGetVal(pIter->pRow->pTSRow, pIter->pTSchema, pIter->i, &pIter->colVal);
...@@ -1084,11 +1084,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch ...@@ -1084,11 +1084,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch
cv.flag = CV_FLAG_VALUE; cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset - sizeof(TSKEY)); void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset);
cv.value.nData = varDataLen(pData); cv.value.nData = varDataLen(pData);
cv.value.pData = varDataVal(pData); cv.value.pData = varDataVal(pData);
} else { } else {
memcpy(&cv.value.val, pRow->data + pTColumn->offset - sizeof(TSKEY), pTColumn->bytes); memcpy(&cv.value.val, pRow->data + pTColumn->offset, pTColumn->bytes);
} }
code = tColDataAppendValue(pColData, &cv); code = tColDataAppendValue(pColData, &cv);
...@@ -1106,11 +1106,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch ...@@ -1106,11 +1106,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch
cv.flag = CV_FLAG_VALUE; cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset - sizeof(TSKEY)); void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset);
cv.value.nData = varDataLen(pData); cv.value.nData = varDataLen(pData);
cv.value.pData = varDataVal(pData); cv.value.pData = varDataVal(pData);
} else { } else {
memcpy(&cv.value.val, pRow->data + pTColumn->offset - sizeof(TSKEY), pTColumn->bytes); memcpy(&cv.value.val, pRow->data + pTColumn->offset, pTColumn->bytes);
} }
code = tColDataAppendValue(pColData, &cv); code = tColDataAppendValue(pColData, &cv);
......
source/libs/parser/src/parInsertUtil.c
浏览文件 @ f58396ea
...@@ -139,8 +139,8 @@ void insSetBoundColumnInfo(SParsedDataColInfo* pColList, SSchema* pSchema, col_i ...@@ -139,8 +139,8 @@ void insSetBoundColumnInfo(SParsedDataColInfo* pColList, SSchema* pSchema, col_i
if (i > 0) { if (i > 0) {
pColList->cols[i].offset = pColList->cols[i - 1].offset + pSchema[i - 1].bytes; pColList->cols[i].offset = pColList->cols[i - 1].offset + pSchema[i - 1].bytes;
pColList->cols[i].toffset = pColList->flen; pColList->cols[i].toffset = pColList->flen;
pColList->flen += TYPE_BYTES[type];
} }
pColList->flen += TYPE_BYTES[type];
switch (type) { switch (type) {
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
pColList->allNullLen += (VARSTR_HEADER_SIZE + CHAR_BYTES); pColList->allNullLen += (VARSTR_HEADER_SIZE + CHAR_BYTES);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或