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

Merge pull request #13375 from taosdata/feat/row_refact 

feat: vnode multi-version 1
This commit is contained in:
Hongze Cheng 2022-06-01 16:52:36 +08:00 committed by GitHub
parent 41132d779a 6f94998e06
commit d8ac914a69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 1173 additions and 2760 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
include/common/tdataformat.h
View File

@ -30,6 +30,7 @@ extern "C" {
typedef struct SSchema SSchema;
typedef struct STColumn STColumn;
typedef struct STSchema STSchema;
typedef struct SValue SValue;
typedef struct SColVal SColVal;
typedef struct STSRow2 STSRow2;
typedef struct STSRowBuilder STSRowBuilder;
@ -40,24 +41,26 @@ typedef struct STag STag;
int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema);
void tTSchemaDestroy(STSchema *pTSchema);
// SColVal
#define ColValNONE ((SColVal){.type = COL_VAL_NONE, .nData = 0, .pData = NULL})
#define ColValNULL ((SColVal){.type = COL_VAL_NULL, .nData = 0, .pData = NULL})
#define ColValDATA(nData, pData) ((SColVal){.type = COL_VAL_DATA, .nData = (nData), .pData = (pData)})
// STSRow2
#define COL_VAL_NONE(CID) ((SColVal){.cid = (CID), .isNone = 1})
#define COL_VAL_NULL(CID) ((SColVal){.cid = (CID), .isNull = 1})
#define COL_VAL_VALUE(CID, V) ((SColVal){.cid = (CID), .value = (V)})
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow);
void tTSRowFree(STSRow2 *pRow);
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray);
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow);
int32_t tTSRowDup(const STSRow2 *pRow, STSRow2 **ppRow);
void tTSRowFree(STSRow2 *pRow);
int32_t tTSRowGet(const STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
// STSRowBuilder
#if 0
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema);
void tTSRowBuilderClear(STSRowBuilder *pBuilder);
void tTSRowBuilderReset(STSRowBuilder *pBuilder);
int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData);
int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow);
#endif
// STag
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
@ -90,7 +93,9 @@ struct STSchema {
#define TSROW_HAS_NONE ((uint8_t)0x1)
#define TSROW_HAS_NULL ((uint8_t)0x2U)
#define TSROW_HAS_VAL ((uint8_t)0x4U)
#define TSROW_KV_ROW ((uint8_t)0x10U)
#define TSROW_KV_SMALL ((uint8_t)0x10U)
#define TSROW_KV_MID ((uint8_t)0x20U)
#define TSROW_KV_BIG ((uint8_t)0x40U)
struct STSRow2 {
TSKEY ts;
uint8_t flags;
@ -113,11 +118,31 @@ struct STSRowBuilder {
STSRow2 row;
};
typedef enum { COL_VAL_NONE = 0, COL_VAL_NULL = 1, COL_VAL_DATA = 2 } EColValT;
struct SValue {
union {
int8_t i8; // TSDB_DATA_TYPE_BOOL||TSDB_DATA_TYPE_TINYINT
uint8_t u8; // TSDB_DATA_TYPE_UTINYINT
int16_t i16; // TSDB_DATA_TYPE_SMALLINT
uint16_t u16; // TSDB_DATA_TYPE_USMALLINT
int32_t i32; // TSDB_DATA_TYPE_INT
uint32_t u32; // TSDB_DATA_TYPE_UINT
int64_t i64; // TSDB_DATA_TYPE_BIGINT
uint64_t u64; // TSDB_DATA_TYPE_UBIGINT
TSKEY ts; // TSDB_DATA_TYPE_TIMESTAMP
float f; // TSDB_DATA_TYPE_FLOAT
double d; // TSDB_DATA_TYPE_DOUBLE
struct {
uint32_t nData;
uint8_t *pData;
};
};
};
struct SColVal {
EColValT type;
uint32_t nData;
uint8_t *pData;
int16_t cid;
int8_t isNone;
int8_t isNull;
SValue value;
};
#pragma pack(push, 1)

20
include/common/tmsg.h
View File

@ -945,7 +945,6 @@ typedef struct {
int64_t timeInFetchQueue;
} SQnodeLoad;
typedef struct {
int32_t sver; // software version
int64_t dnodeVer; // dnode table version in sdb
@ -1977,7 +1976,7 @@ typedef struct {
int8_t killConnection;
int8_t align[3];
SEpSet epSet;
SArray *pQnodeList;
SArray* pQnodeList;
} SQueryHbRspBasic;
typedef struct {
@ -2663,6 +2662,23 @@ typedef struct {
int32_t tEncodeSVSubmitReq(SEncoder* pCoder, const SVSubmitReq* pReq);
int32_t tDecodeSVSubmitReq(SDecoder* pCoder, SVSubmitReq* pReq);
// TDMT_VND_DELETE
typedef struct {
TSKEY sKey;
TSKEY eKey;
// super table
char* stbName;
// child/normal
char* tbName;
} SVDeleteReq;
typedef struct {
int32_t code;
// TODO
} SVDeleteRsp;
#pragma pack(pop)
#ifdef __cplusplus

2
include/common/tmsgdef.h
View File

@ -225,6 +225,8 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_VNODE, "vnode-alter-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_COMPACT_VNODE, "vnode-compact-vnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "vnode-delete-data", SVDeleteReq, SVDeleteRsp)
// Requests handled by QNODE
TD_NEW_MSG_SEG(TDMT_QND_MSG)

189
include/util/tencode.h
View File

@ -461,64 +461,153 @@ static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) {
}
// ===========================================
#define tPutV(p, v) \
int32_t n = 0; \
for (;;) { \
if (v <= 0x7f) { \
if (p) p[n] = v; \
n++; \
break; \
} \
if (p) p[n] = (v & 0x7f) | 0x80; \
n++; \
v >>= 7; \
} \
return n;
#define tPutV(p, v) \
do { \
int32_t n = 0; \
for (;;) { \
if (v <= 0x7f) { \
if (p) p[n] = v; \
n++; \
break; \
} \
if (p) p[n] = (v & 0x7f) | 0x80; \
n++; \
v >>= 7; \
} \
return n; \
} while (0)
#define tGetV(p, v) \
int32_t n = 0; \
if (v) *v = 0; \
for (;;) { \
if (p[n] <= 0x7f) { \
if (v) (*v) |= (p[n] << (7 * n)); \
n++; \
break; \
} \
if (v) (*v) |= ((p[n] & 0x7f) << (7 * n)); \
n++; \
} \
return n;
#define tGetV(p, v) \
do { \
int32_t n = 0; \
if (v) *v = 0; \
for (;;) { \
if (p[n] <= 0x7f) { \
if (v) (*v) |= (p[n] << (7 * n)); \
n++; \
break; \
} \
if (v) (*v) |= ((p[n] & 0x7f) << (7 * n)); \
n++; \
} \
return n; \
} while (0)
// PUT
static FORCE_INLINE int32_t tPutU8(uint8_t* p, uint8_t v) {
if (p) ((uint8_t*)p)[0] = v;
return sizeof(uint8_t);
}
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 tPutU16v(uint8_t* p, uint16_t v) { tPutV(p, v) }
static FORCE_INLINE int32_t tPutFloat(uint8_t* p, float f) {
union {
uint32_t ui;
float f;
} v;
v.f = f;
return tPutU32(p, v.ui);
}
static FORCE_INLINE int32_t tPutDouble(uint8_t* p, double d) {
union {
uint64_t ui;
double d;
} v;
v.d = d;
return tPutU64(p, v.ui);
}
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) { return tPutU16v(p, ZIGZAGE(int16_t, v)); }
static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v) { tPutV(p, v) }
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) { return tPutU32v(p, ZIGZAGE(int32_t, v)); }
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) { return tPutU64v(p, ZIGZAGE(int64_t, v)); }
// GET
static FORCE_INLINE int32_t tGetU8(uint8_t* p, uint8_t* v) {
if (v) *v = ((uint8_t*)p)[0];
return sizeof(uint8_t);
}
static FORCE_INLINE int32_t tGetI8(uint8_t* p, int8_t* v) {
if (v) *v = ((int8_t*)p)[0];
return sizeof(int8_t);
}
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) { tGetV(p, v) }
static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v) { tGetV(p, v); }
static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
int32_t n;
@ -530,7 +619,7 @@ static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
return n;
}
static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) { tGetV(p, v) }
static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) { tGetV(p, v); }
static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
int32_t n;
@ -542,6 +631,46 @@ static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
return n;
}
static FORCE_INLINE int32_t tGetU64v(uint8_t* p, uint64_t* v) { tGetV(p, v); }
static FORCE_INLINE int32_t tGetI64v(uint8_t* p, int64_t* v) {
int32_t n;
uint64_t tv;
n = tGetU64v(p, &tv);
if (v) *v = ZIGZAGD(int64_t, tv);
return n;
}
static FORCE_INLINE int32_t tGetFloat(uint8_t* p, float* f) {
int32_t n = 0;
union {
uint32_t ui;
float f;
} v;
n = tGetU32(p, &v.ui);
*f = v.f;
return n;
}
static FORCE_INLINE int32_t tGetDouble(uint8_t* p, double* d) {
int32_t n = 0;
union {
uint64_t ui;
double d;
} v;
n = tGetU64(p, &v.ui);
*d = v.d;
return n;
}
// =====================
static FORCE_INLINE int32_t tPutBinary(uint8_t* p, uint8_t* pData, uint32_t nData) {
int n = 0;

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

@ -21,15 +21,10 @@
static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson);
typedef struct SKVIdx {
int32_t cid;
int32_t offset;
} SKVIdx;
#pragma pack(push, 1)
typedef struct {
int16_t nCols;
SKVIdx idx[];
uint8_t idx[];
} STSKVRow;
#pragma pack(pop)
@ -43,7 +38,507 @@ typedef struct {
static FORCE_INLINE int tSKVIdxCmprFn(const void *p1, const void *p2);
// STSRow2
// SValue
static FORCE_INLINE int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) {
int32_t n = 0;
if (IS_VAR_DATA_TYPE(type)) {
n += tPutBinary(p ? p + n : p, pValue->pData, pValue->nData);
} else {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
n += tPutI8(p ? p + n : p, pValue->i8 ? 1 : 0);
break;
case TSDB_DATA_TYPE_TINYINT:
n += tPutI8(p ? p + n : p, pValue->i8);
break;
case TSDB_DATA_TYPE_SMALLINT:
n += tPutI16(p ? p + n : p, pValue->i16);
break;
case TSDB_DATA_TYPE_INT:
n += tPutI32(p ? p + n : p, pValue->i32);
break;
case TSDB_DATA_TYPE_BIGINT:
n += tPutI64(p ? p + n : p, pValue->i64);
break;
case TSDB_DATA_TYPE_FLOAT:
n += tPutFloat(p ? p + n : p, pValue->f);
break;
case TSDB_DATA_TYPE_DOUBLE:
n += tPutDouble(p ? p + n : p, pValue->d);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
n += tPutI64(p ? p + n : p, pValue->ts);
break;
case TSDB_DATA_TYPE_UTINYINT:
n += tPutU8(p ? p + n : p, pValue->u8);
break;
case TSDB_DATA_TYPE_USMALLINT:
n += tPutU16(p ? p + n : p, pValue->u16);
break;
case TSDB_DATA_TYPE_UINT:
n += tPutU32(p ? p + n : p, pValue->u32);
break;
case TSDB_DATA_TYPE_UBIGINT:
n += tPutU64(p ? p + n : p, pValue->u64);
break;
default:
ASSERT(0);
}
}
return n;
}
static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
int32_t n = 0;
if (IS_VAR_DATA_TYPE(type)) {
n += tGetBinary(p, &pValue->pData, pValue ? &pValue->nData : NULL);
} else {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
n += tGetI8(p, &pValue->i8);
break;
case TSDB_DATA_TYPE_TINYINT:
n += tGetI8(p, &pValue->i8);
break;
case TSDB_DATA_TYPE_SMALLINT:
n += tGetI16(p, &pValue->i16);
break;
case TSDB_DATA_TYPE_INT:
n += tGetI32(p, &pValue->i32);
break;
case TSDB_DATA_TYPE_BIGINT:
n += tGetI64(p, &pValue->i64);
break;
case TSDB_DATA_TYPE_FLOAT:
n += tGetFloat(p, &pValue->f);
break;
case TSDB_DATA_TYPE_DOUBLE:
n += tGetDouble(p, &pValue->d);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
n += tGetI64(p, &pValue->ts);
break;
case TSDB_DATA_TYPE_UTINYINT:
n += tGetU8(p, &pValue->u8);
break;
case TSDB_DATA_TYPE_USMALLINT:
n += tGetU16(p, &pValue->u16);
break;
case TSDB_DATA_TYPE_UINT:
n += tGetU32(p, &pValue->u32);
break;
case TSDB_DATA_TYPE_UBIGINT:
n += tGetU64(p, &pValue->u64);
break;
default:
ASSERT(0);
}
}
return n;
}
// STSRow2 ========================================================================
static void tTupleTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
int32_t nColVal = taosArrayGetSize(pArray);
STColumn *pTColumn;
SColVal *pColVal;
ASSERT(nColVal > 0);
pRow->sver = pTSchema->version;
// ts
pTColumn = &pTSchema->columns[0];
pColVal = (SColVal *)taosArrayGet(pArray, 0);
ASSERT(pTColumn->colId == 0 && pColVal->cid == 0);
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
pRow->ts = pColVal->value.ts;
// other fields
int32_t iColVal = 1;
int32_t bidx;
uint32_t nv = 0;
uint8_t *pb = NULL;
uint8_t *pf = NULL;
uint8_t *pv = NULL;
uint8_t flags = 0;
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
bidx = iColumn - 1;
pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else {
pColVal = NULL;
}
if (pColVal) {
if (pColVal->cid == pTColumn->colId) {
iColVal++;
if (pColVal->isNone) {
goto _set_none;
} else if (pColVal->isNull) {
goto _set_null;
} else {
goto _set_value;
}
} else if (pColVal->cid > pTColumn->colId) {
goto _set_none;
} else {
ASSERT(0);
}
} else {
goto _set_none;
}
_set_none:
flags |= TSROW_HAS_NONE;
// SET_BIT2(pb, bidx, 0); (todo)
continue;
_set_null:
flags != TSROW_HAS_NULL;
// SET_BIT2(pb, bidx, 1); (todo)
continue;
_set_value:
flags != TSROW_HAS_VAL;
// SET_BIT2(pb, bidx, 2); (todo)
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 1);
} else {
// tPutColVal(pf ? pf + pTColumn->offset : pf, pColVal, pTColumn->type, 1);
}
continue;
}
ASSERT(flags);
switch (flags & 0xf) {
case TSROW_HAS_NONE:
case TSROW_HAS_NULL:
pRow->nData = 0;
break;
case TSROW_HAS_VAL:
pRow->nData = pTSchema->flen + nv;
break;
case TSROW_HAS_NULL | TSROW_HAS_NONE:
pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1);
break;
case TSROW_HAS_VAL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NULL:
pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv;
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
pRow->nData = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv;
break;
default:
break;
}
}
static void tMapTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
int32_t nColVal = taosArrayGetSize(pArray);
STColumn *pTColumn;
SColVal *pColVal;
ASSERT(nColVal > 0);
pRow->sver = pTSchema->version;
// ts
pTColumn = &pTSchema->columns[0];
pColVal = (SColVal *)taosArrayGet(pArray, 0);
ASSERT(pTColumn->colId == 0 && pColVal->cid == 0);
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
pRow->ts = pColVal->value.ts;
// other fields
int32_t iColVal = 1;
uint32_t nv = 0;
uint8_t *pv = NULL;
uint8_t *pidx = NULL;
uint8_t flags = 0;
int16_t nCol = 0;
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else {
pColVal = NULL;
}
if (pColVal) {
if (pColVal->cid == pTColumn->colId) {
iColVal++;
if (pColVal->isNone) {
goto _set_none;
} else if (pColVal->isNull) {
goto _set_null;
} else {
goto _set_value;
}
} else if (pColVal->cid > pTColumn->colId) {
goto _set_none;
} else {
ASSERT(0);
}
} else {
goto _set_none;
}
_set_none:
flags |= TSROW_HAS_NONE;
continue;
_set_null:
flags != TSROW_HAS_NULL;
pidx[nCol++] = nv;
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0);
continue;
_set_value:
flags != TSROW_HAS_VAL;
pidx[nCol++] = nv;
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0);
continue;
}
if (nv <= UINT8_MAX) {
// small
} else if (nv <= UINT16_MAX) {
// mid
} else {
// large
}
}
// try-decide-build
int32_t tTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
int32_t code = 0;
STSRow2 rowT = {0};
STSRow2 rowM = {0};
// try
tTupleTSRowNew(pArray, pTSchema, &rowT);
tMapTSRowNew(pArray, pTSchema, &rowM);
// decide & build
if (rowT.nData <= rowM.nData) {
tTupleTSRowNew(pArray, pTSchema, &rowT);
} else {
tMapTSRowNew(pArray, pTSchema, &rowM);
}
return code;
}
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow) {
int32_t code = 0;
(*ppRow) = (STSRow2 *)taosMemoryMalloc(sizeof(**ppRow));
if (*ppRow == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
**ppRow = *pRow;
(*ppRow)->pData = NULL;
if (pRow->nData) {
(*ppRow)->pData = taosMemoryMalloc(pRow->nData);
if ((*ppRow)->pData == NULL) {
taosMemoryFree(*ppRow);
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
memcpy((*ppRow)->pData, pRow->pData, pRow->nData);
}
_exit:
return code;
}
void tTSRowFree(STSRow2 *pRow) {
if (pRow) {
if (pRow->pData) taosMemoryFree(pRow->pData);
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(flags);
ASSERT(pRow->sver == pTSchema->version);
if (iCol == 0) {
value.ts = pRow->ts;
goto _return_value;
}
if (flags == TSROW_HAS_NONE) {
goto _return_none;
} else if (flags == TSROW_HAS_NONE) {
goto _return_null;
}
ASSERT(pRow->nData && pRow->pData);
if (isTuple) {
uint8_t *pb = pRow->pData;
uint8_t *pf = NULL;
uint8_t *pv = NULL;
uint8_t *p;
uint8_t b;
// bit
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 {
ASSERT(0);
}
while (lidx <= ridx) {
midx = (lidx + ridx) / 2;
if (pRow->flags & TSROW_KV_SMALL) {
n = ((uint8_t *)pRowK->idx)[midx];
} else if (pRow->flags & TSROW_KV_MID) {
n = ((uint16_t *)pRowK->idx)[midx];
} else {
n = ((uint32_t *)pRowK->idx)[midx];
}
n += tGetI16v(p + n, &cid);
if (TABS(cid) == pTColumn->colId) {
if (cid < 0) {
goto _return_null;
} else {
n += tGetValue(p + n, &value, pTColumn->type);
goto _return_value;
}
return;
} else if (TABS(cid) > pTColumn->colId) {
ridx = midx - 1;
} else {
lidx = midx + 1;
}
}
// not found, return NONE
goto _return_none;
}
_return_none:
*pColVal = COL_VAL_NONE(pTColumn->colId);
return;
_return_null:
*pColVal = COL_VAL_NULL(pTColumn->colId);
return;
_return_value:
*pColVal = COL_VAL_VALUE(pTColumn->colId, value);
return;
}
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) {
int32_t code = 0;
SColVal cv;
(*ppArray) = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal));
if (*ppArray == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; iColumn++) {
tTSRowGet(pRow, pTSchema, iColumn, &cv);
taosArrayPush(*ppArray, &cv);
}
_exit:
return code;
}
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
int32_t n = 0;
@ -56,8 +551,11 @@ int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
switch (pRow->flags & 0xf) {
case TSROW_HAS_NONE:
case TSROW_HAS_NULL:
ASSERT(pRow->nData == 0);
ASSERT(pRow->pData == NULL);
break;
default:
ASSERT(pRow->nData && pRow->pData);
n += tPutBinary(p ? p + n : p, pRow->pData, pRow->nData);
break;
}
@ -67,149 +565,26 @@ int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow) {
int32_t n = 0;
uint8_t flags;
n += tGetI64(p + n, pRow ? &pRow->ts : NULL);
n += tGetI8(p + n, pRow ? &pRow->flags : &flags);
n += tGetI32v(p + n, pRow ? &pRow->sver : NULL);
n += tGetI64(p + n, &pRow->ts);
n += tGetI8(p + n, &pRow->flags);
n += tGetI32v(p + n, &pRow->sver);
if (pRow) flags = pRow->flags;
switch (flags & 0xf) {
ASSERT(pRow->flags);
switch (pRow->flags & 0xf) {
case TSROW_HAS_NONE:
case TSROW_HAS_NULL:
pRow->nData = 0;
pRow->pData = NULL;
break;
default:
n += tGetBinary(p + n, pRow ? &pRow->pData : NULL, pRow ? &pRow->nData : NULL);
n += tGetBinary(p + n, &pRow->pData, &pRow->nData);
break;
}
return n;
}
int32_t tTSRowDup(const STSRow2 *pRow, STSRow2 **ppRow) {
(*ppRow) = taosMemoryMalloc(sizeof(*pRow) + pRow->nData);
if (*ppRow == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
(*ppRow)->ts = pRow->ts;
(*ppRow)->flags = pRow->flags;
(*ppRow)->sver = pRow->sver;
(*ppRow)->nData = pRow->nData;
if (pRow->nData) {
(*ppRow)->pData = (uint8_t *)(&(*ppRow)[1]);
memcpy((*ppRow)->pData, pRow->pData, pRow->nData);
} else {
(*ppRow)->pData = NULL;
}
return 0;
}
void tTSRowFree(STSRow2 *pRow) {
if (pRow) taosMemoryFree(pRow);
}
int32_t tTSRowGet(const STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
uint32_t n;
uint8_t *p;
uint8_t v;
int32_t bidx = iCol - 1;
STColumn *pTColumn = &pTSchema->columns[iCol];
STSKVRow *pTSKVRow;
SKVIdx *pKVIdx;
ASSERT(iCol != 0);
ASSERT(pTColumn->colId != 0);
ASSERT((pRow->flags & 0xf) != 0);
switch (pRow->flags & 0xf) {
case TSROW_HAS_NONE:
*pColVal = ColValNONE;
return 0;
case TSROW_HAS_NULL:
*pColVal = ColValNULL;
return 0;
}
if (TSROW_IS_KV_ROW(pRow)) {
ASSERT((pRow->flags & 0xf) != TSROW_HAS_VAL);
pTSKVRow = (STSKVRow *)pRow->pData;
pKVIdx =
bsearch(&((SKVIdx){.cid = pTColumn->colId}), pTSKVRow->idx, pTSKVRow->nCols, sizeof(SKVIdx), tSKVIdxCmprFn);
if (pKVIdx == NULL) {
*pColVal = ColValNONE;
} else if (pKVIdx->offset < 0) {
*pColVal = ColValNULL;
} else {
p = pRow->pData + sizeof(STSKVRow) + sizeof(SKVIdx) * pTSKVRow->nCols + pKVIdx->offset;
pColVal->type = COL_VAL_DATA;
tGetBinary(p, &pColVal->pData, &pColVal->nData);
}
} else {
// get bitmap
p = pRow->pData;
switch (pRow->flags & 0xf) {
case TSROW_HAS_NULL | TSROW_HAS_NONE:
v = GET_BIT1(p, bidx);
if (v == 0) {
*pColVal = ColValNONE;
} else {
*pColVal = ColValNULL;
}
return 0;
case TSROW_HAS_VAL | TSROW_HAS_NONE:
v = GET_BIT1(p, bidx);
if (v == 1) {
p = p + BIT1_SIZE(pTSchema->numOfCols - 1);
break;
} else {
*pColVal = ColValNONE;
return 0;
}
case TSROW_HAS_VAL | TSROW_HAS_NULL:
v = GET_BIT1(p, bidx);
if (v == 1) {
p = p + BIT1_SIZE(pTSchema->numOfCols - 1);
break;
} else {
*pColVal = ColValNULL;
return 0;
}
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
v = GET_BIT2(p, bidx);
if (v == 0) {
*pColVal = ColValNONE;
return 0;
} else if (v == 1) {
*pColVal = ColValNULL;
return 0;
} else if (v == 2) {
p = p + BIT2_SIZE(pTSchema->numOfCols - 1);
break;
} else {
ASSERT(0);
}
default:
break;
}
// get real value
p = p + pTColumn->offset;
pColVal->type = COL_VAL_DATA;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
tGetBinary(p + pTSchema->flen + *(int32_t *)p, &pColVal->pData, &pColVal->nData);
} else {
pColVal->pData = p;
pColVal->nData = pTColumn->bytes;
}
}
return 0;
}
// STSchema
int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t ncols, STSchema **ppTSchema) {
*ppTSchema = (STSchema *)taosMemoryMalloc(sizeof(STSchema) + sizeof(STColumn) * ncols);
@ -251,6 +626,7 @@ void tTSchemaDestroy(STSchema *pTSchema) {
}
// STSRowBuilder
#if 0
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema) {
if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1;
@ -508,6 +884,7 @@ int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) {
return 0;
}
#endif
static int tTagValCmprFn(const void *p1, const void *p2) {
if (((STagVal *)p1)->cid < ((STagVal *)p2)->cid) {
@ -622,9 +999,9 @@ void debugPrintSTag(STag *pTag, const char *tag, int32_t ln) {
}
printf("%s:%d loop[%d-%d] offset=%d\n", __func__, __LINE__, (int32_t)pTag->nTag, (int32_t)n, (int32_t)offset);
tGetTagVal(p + offset, &tagVal, isJson);
if(IS_VAR_DATA_TYPE(tagVal.type)){
if (IS_VAR_DATA_TYPE(tagVal.type)) {
debugPrintTagVal(tagVal.type, tagVal.pData, tagVal.nData, __func__, __LINE__);
}else{
} else {
debugPrintTagVal(tagVal.type, &tagVal.i64, tDataTypes[tagVal.type].bytes, __func__, __LINE__);
}
}
@ -650,7 +1027,7 @@ static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
} else {
p = p ? p + n : p;
n += tDataTypes[pTagVal->type].bytes;
if(p) memcpy(p, &(pTagVal->i64), tDataTypes[pTagVal->type].bytes);
if (p) memcpy(p, &(pTagVal->i64), tDataTypes[pTagVal->type].bytes);
}
return n;
@ -750,21 +1127,21 @@ void tTagFree(STag *pTag) {
if (pTag) taosMemoryFree(pTag);
}
char *tTagValToData(const STagVal *value, bool isJson){
if(!value) return NULL;
char *data = NULL;
char *tTagValToData(const STagVal *value, bool isJson) {
if (!value) return NULL;
char *data = NULL;
int8_t typeBytes = 0;
if (isJson) {
typeBytes = CHAR_BYTES;
}
if(IS_VAR_DATA_TYPE(value->type)){
if (IS_VAR_DATA_TYPE(value->type)) {
data = taosMemoryCalloc(1, typeBytes + VARSTR_HEADER_SIZE + value->nData);
if(data == NULL) return NULL;
if(isJson) *data = value->type;
if (data == NULL) return NULL;
if (isJson) *data = value->type;
varDataLen(data + typeBytes) = value->nData;
memcpy(varDataVal(data + typeBytes), value->pData, value->nData);
}else{
data = ((char*)&(value->i64)) - typeBytes; // json with type
} else {
data = ((char *)&(value->i64)) - typeBytes; // json with type
}
return data;

2
source/dnode/vnode/CMakeLists.txt
View File

@ -35,7 +35,6 @@ target_sources(
"src/sma/smaTimeRange.c"
# tsdb
# "src/tsdb/tsdbTDBImpl.c"
"src/tsdb/tsdbCommit.c"
"src/tsdb/tsdbCommit2.c"
"src/tsdb/tsdbFile.c"
@ -45,7 +44,6 @@ target_sources(
"src/tsdb/tsdbMemTable2.c"
"src/tsdb/tsdbRead.c"
"src/tsdb/tsdbReadImpl.c"
# "src/tsdb/tsdbSma.c"
"src/tsdb/tsdbWrite.c"
"src/tsdb/tsdbSnapshot.c"

51
source/dnode/vnode/src/inc/tsdb.h
View File

@ -32,14 +32,27 @@ extern "C" {
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TSDB ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on
typedef struct TSDBROW TSDBROW;
typedef struct TSDBKEY TSDBKEY;
typedef struct SDelOp SDelOp;
static int tsdbKeyCmprFn(const void *p1, const void *p2);
// tsdbMemTable2.c ==============================================================================================
typedef struct SMemTable SMemTable;
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable);
void tsdbMemTableDestroy2(SMemTable *pMemTable);
// tsdbMemTable ================
typedef struct STsdbRow STsdbRow;
typedef struct STbData STbData;
typedef struct STsdbMemTable STsdbMemTable;
typedef struct SMergeInfo SMergeInfo;
typedef struct STable STable;
int tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable);
void tsdbMemTableDestroy(STsdbMemTable *pMemTable);
int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
@ -845,6 +858,42 @@ static FORCE_INLINE int tsdbUnLockFS(STsdbFS *pFs) {
return 0;
}
struct TSDBROW {
int64_t version;
STSRow2 tsRow;
};
struct TSDBKEY {
int64_t version;
TSKEY ts;
};
struct SDelOp {
int64_t version;
TSKEY sKey; // included
TSKEY eKey; // included
SDelOp *pNext;
};
static FORCE_INLINE int tsdbKeyCmprFn(const void *p1, const void *p2) {
TSDBKEY *pKey1 = (TSDBKEY *)p1;
TSDBKEY *pKey2 = (TSDBKEY *)p2;
if (pKey1->ts < pKey2->ts) {
return -1;
} else if (pKey1->ts > pKey2->ts) {
return 1;
}
if (pKey1->version < pKey2->version) {
return -1;
} else if (pKey1->version > pKey2->version) {
return 1;
}
return 0;
}
#endif
#ifdef __cplusplus

57
source/dnode/vnode/src/meta/metaTable.c
View File

@ -31,9 +31,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
int vLen = 0;
const void *pKey = NULL;
const void *pVal = NULL;
void * pBuf = NULL;
void *pBuf = NULL;
int32_t szBuf = 0;
void * p = NULL;
void *p = NULL;
SMetaReader mr = {0};
// validate req
@ -87,7 +87,7 @@ int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
}
// drop all child tables
TBC * pCtbIdxc = NULL;
TBC *pCtbIdxc = NULL;
SArray *pArray = taosArrayInit(8, sizeof(tb_uid_t));
tdbTbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
@ -142,8 +142,8 @@ _exit:
int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
SMetaEntry oStbEntry = {0};
SMetaEntry nStbEntry = {0};
TBC * pUidIdxc = NULL;
TBC * pTbDbc = NULL;
TBC *pUidIdxc = NULL;
TBC *pTbDbc = NULL;
const void *pData;
int nData;
int64_t oversion;
@ -262,7 +262,7 @@ _err:
}
int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
void * pData = NULL;
void *pData = NULL;
int nData = 0;
int rc = 0;
tb_uid_t uid;
@ -288,7 +288,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
}
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
void * pData = NULL;
void *pData = NULL;
int nData = 0;
int rc = 0;
int64_t version;
@ -324,14 +324,14 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
}
static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
void * pVal = NULL;
void *pVal = NULL;
int nVal = 0;
const void * pData = NULL;
const void *pData = NULL;
int nData = 0;
int ret = 0;
tb_uid_t uid;
int64_t oversion;
SSchema * pColumn = NULL;
SSchema *pColumn = NULL;
SMetaEntry entry = {0};
SSchemaWrapper *pSchema;
int c;
@ -479,7 +479,7 @@ _err:
static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
SMetaEntry ctbEntry = {0};
SMetaEntry stbEntry = {0};
void * pVal = NULL;
void *pVal = NULL;
int nVal = 0;
int ret;
int c;
@ -510,7 +510,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
oversion = *(int64_t *)pData;
// search table.db
TBC * pTbDbc = NULL;
TBC *pTbDbc = NULL;
SDecoder dc1 = {0};
SDecoder dc2 = {0};
@ -534,7 +534,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
metaDecodeEntry(&dc2, &stbEntry);
SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
SSchema * pColumn = NULL;
SSchema *pColumn = NULL;
int32_t iCol = 0;
for (;;) {
pColumn = NULL;
@ -579,7 +579,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
if (IS_VAR_DATA_TYPE(pCol->type)) {
val.pData = pAlterTbReq->pTagVal;
val.nData = pAlterTbReq->nTagVal;
}else{
} else {
memcpy(&val.i64, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
}
taosArrayPush(pTagArray, &val);
@ -649,8 +649,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq) {
static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
STbDbKey tbDbKey;
void * pKey = NULL;
void * pVal = NULL;
void *pKey = NULL;
void *pVal = NULL;
int kLen = 0;
int vLen = 0;
SEncoder coder = {0};
@ -732,7 +732,7 @@ static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) {
}
int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_t nTagData, int8_t type, tb_uid_t uid,
STagIdxKey **ppTagIdxKey, int32_t *nTagIdxKey) {
STagIdxKey **ppTagIdxKey, int32_t *nTagIdxKey) {
// int32_t nTagData = 0;
// if (pTagData) {
@ -765,11 +765,11 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) {
}
static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
void * pData = NULL;
void *pData = NULL;
int nData = 0;
STbDbKey tbDbKey = {0};
SMetaEntry stbEntry = {0};
STagIdxKey * pTagIdxKey = NULL;
STagIdxKey *pTagIdxKey = NULL;
int32_t nTagIdxKey;
const SSchema *pTagColumn; // = &stbEntry.stbEntry.schema.pSchema[0];
const void *pTagData = NULL; //
@ -788,21 +788,20 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
pTagColumn = &stbEntry.stbEntry.schemaTag.pSchema[0];
STagVal tagVal = {.cid = pTagColumn->colId};
if(pTagColumn->type != TSDB_DATA_TYPE_JSON){
if (pTagColumn->type != TSDB_DATA_TYPE_JSON) {
tTagGet((const STag *)pCtbEntry->ctbEntry.pTags, &tagVal);
if(IS_VAR_DATA_TYPE(pTagColumn->type)){
if (IS_VAR_DATA_TYPE(pTagColumn->type)) {
pTagData = tagVal.pData;
nTagData = (int32_t)tagVal.nData;
}else{
} else {
pTagData = &(tagVal.i64);
nTagData = tDataTypes[pTagColumn->type].bytes;
}
}else{
//pTagData = pCtbEntry->ctbEntry.pTags;
//nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len;
} else {
// pTagData = pCtbEntry->ctbEntry.pTags;
// nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len;
}
// update tag index
#ifdef USE_INVERTED_INDEX
tb_uid_t suid = pCtbEntry->ctbEntry.suid;
@ -816,8 +815,8 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
int ret = indexPut((SIndex *)pMeta->pTagIvtIdx, tmGroup, tuid);
indexMultiTermDestroy(tmGroup);
#else
if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type, pCtbEntry->uid,
&pTagIdxKey, &nTagIdxKey) < 0) {
if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type,
pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) {
return -1;
}
tdbTbInsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn);
@ -830,7 +829,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
SEncoder coder = {0};
void * pVal = NULL;
void *pVal = NULL;
int vLen = 0;
int rcode = 0;
SSkmDbKey skmDbKey = {0};

2
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -238,7 +238,7 @@ static void tsdbStartCommit(STsdb *pRepo) {
static void tsdbEndCommit(STsdb *pTsdb, int eno) {
tsdbEndFSTxn(pTsdb);
tsdbMemTableDestroy(pTsdb, pTsdb->imem);
tsdbMemTableDestroy(pTsdb->imem);
pTsdb->imem = NULL;
tsdbInfo("vgId:%d commit over, %s", REPO_ID(pTsdb), (eno == TSDB_CODE_SUCCESS) ? "succeed" : "failed");
}

14
source/dnode/vnode/src/tsdb/tsdbDelete.c Normal file
View File

@ -0,0 +1,14 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/

67
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -60,7 +60,7 @@ int tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable) {
return 0;
}
void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable) {
void tsdbMemTableDestroy(STsdbMemTable *pMemTable) {
if (pMemTable) {
taosHashCleanup(pMemTable->pHashIdx);
SSkipListIterator *pIter = tSkipListCreateIter(pMemTable->pSlIdx);
@ -142,69 +142,6 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter
} else {
fKey = tdGetKey(filterKeys[filterIter]);
}
#if 0
} else if (fKey > rowKey) {
if (isRowDel) {
pMergeInfo->rowsDeleteFailed++;
} else {
if (pMergeInfo->rowsInserted - pMergeInfo->rowsDeleteSucceed >= maxRowsToRead) break;
if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
pMergeInfo->rowsInserted++;
pMergeInfo->nOperations++;
pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey);
pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey);
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
}
tSkipListIterNext(pIter);
row = tsdbNextIterRow(pIter);
if (row == NULL || TD_ROW_KEY(row) > maxKey) {
rowKey = INT64_MAX;
isRowDel = false;
} else {
rowKey = TD_ROW_KEY(row);
isRowDel = TD_ROW_IS_DELETED(row);
}
} else {
if (isRowDel) {
ASSERT(!keepDup);
if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
pMergeInfo->rowsDeleteSucceed++;
pMergeInfo->nOperations++;
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
} else {
if (keepDup) {
if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
pMergeInfo->rowsUpdated++;
pMergeInfo->nOperations++;
pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey);
pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey);
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
} else {
pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, fKey);
pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, fKey);
}
}
tSkipListIterNext(pIter);
row = tsdbNextIterRow(pIter);
if (row == NULL || TD_ROW_KEY(row) > maxKey) {
rowKey = INT64_MAX;
isRowDel = false;
} else {
rowKey = TD_ROW_KEY(row);
isRowDel = TD_ROW_IS_DELETED(row);
}
filterIter++;
if (filterIter >= nFilterKeys) {
fKey = INT64_MAX;
} else {
fKey = tdGetKey(filterKeys[filterIter]);
}
}
#endif
#if 1
} else if (fKey > rowKey) {
if (isRowDel) {
@ -321,7 +258,7 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
return -1;
}
if(pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);
if (pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);
if (mr.me.type == TSDB_NORMAL_TABLE) {
sverNew = mr.me.ntbEntry.schemaRow.version;

386
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
View File

@ -15,52 +15,308 @@
#include "tsdb.h"
typedef struct SMemTable SMemTable;
typedef struct SMemData SMemData;
typedef struct SMemSkipList SMemSkipList;
typedef struct SMemSkipListNode SMemSkipListNode;
typedef struct SMemSkipListCurosr SMemSkipListCurosr;
#define SL_MAX_LEVEL 5
struct SMemTable {
STsdb *pTsdb;
TSKEY minKey;
TSKEY maxKey;
int64_t minVer;
int64_t maxVer;
int64_t nRows;
int32_t nHash;
int32_t nBucket;
SMemData **pBuckets;
SMemSkipListCurosr *pSlc;
};
typedef struct SMemData SMemData;
typedef struct SMemSkipList SMemSkipList;
typedef struct SMemSkipListNode SMemSkipListNode;
struct SMemSkipListNode {
int8_t level;
SMemSkipListNode *forwards[1]; // Windows does not allow 0
SMemSkipListNode *forwards[0];
};
struct SMemSkipList {
uint32_t seed;
int8_t maxLevel;
int8_t level;
int32_t size;
SMemSkipListNode pHead[1]; // Windows does not allow 0
uint32_t seed;
int32_t size;
int8_t maxLevel;
int8_t level;
SMemSkipListNode *pHead;
SMemSkipListNode *pTail;
};
struct SMemData {
SMemData *pHashNext;
tb_uid_t suid;
tb_uid_t uid;
TSKEY minKey;
TSKEY maxKey;
int64_t minVer;
int64_t maxVer;
int64_t nRows;
TSDBKEY minKey;
TSDBKEY maxKey;
SDelOp *delOpHead;
SDelOp *delOpTail;
SMemSkipList sl;
};
struct SMemTable {
STsdb *pTsdb;
int32_t nRef;
TSDBKEY minKey;
TSDBKEY maxKey;
int64_t nRows;
SArray *pArray; // SArray<SMemData>
};
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
#define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level))
#define SL_HEAD_FORWARD(sl, l) SL_NODE_FORWARD((sl)->pHead, l)
#define SL_TAIL_BACKWARD(sl, l) SL_NODE_FORWARD((sl)->pTail, l)
static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData);
static int memDataPCmprFn(const void *p1, const void *p2);
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow);
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
// SMemTable ==============================================
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable) {
int32_t code = 0;
SMemTable *pMemTable = NULL;
pMemTable = (SMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable));
if (pMemTable == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pMemTable->pTsdb = pTsdb;
pMemTable->nRef = 1;
pMemTable->minKey = (TSDBKEY){.version = INT64_MAX, .ts = TSKEY_MAX};
pMemTable->maxKey = (TSDBKEY){.version = -1, .ts = TSKEY_MIN};
pMemTable->nRows = 0;
pMemTable->pArray = taosArrayInit(512, sizeof(SMemData *));
if (pMemTable->pArray == NULL) {
taosMemoryFree(pMemTable);
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*ppMemTable = pMemTable;
return code;
_err:
*ppMemTable = NULL;
return code;
}
void tsdbMemTableDestroy2(SMemTable *pMemTable) {
taosArrayDestroyEx(pMemTable->pArray, NULL /*TODO*/);
taosMemoryFree(pMemTable);
}
int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmitBlk) {
int32_t code = 0;
SMemTable *pMemTable = (SMemTable *)pTsdb->mem; // TODO
SMemData *pMemData;
TSDBROW row = {.version = version};
ASSERT(pMemTable);
{
// check if table exists (todo)
}
code = tsdbGetOrCreateMemData(pMemTable, pSubmitBlk->suid, pSubmitBlk->uid, &pMemData);
if (code) {
tsdbError("vgId:%d failed to create/get table data since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
goto _err;
}
// do insert
int32_t nt;
uint8_t *pt;
int32_t n = 0;
uint8_t *p = pSubmitBlk->pData;
SVBufPool *pPool = pTsdb->pVnode->inUse;
int8_t level;
SMemSkipListNode *pNode;
while (n < pSubmitBlk->nData) {
nt = tGetTSRow(p + n, &row.tsRow);
n += nt;
ASSERT(n <= pSubmitBlk->nData);
// build the node
level = tsdbMemSkipListRandLevel(&pMemData->sl);
pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level) + nt + sizeof(version));
if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pNode->level = level;
tPutTSDBRow((uint8_t *)SL_NODE_DATA(pNode), &row);
// put the node (todo)
// set info
if (tsdbKeyCmprFn(&row, &pMemData->minKey) < 0) pMemData->minKey = *(TSDBKEY *)&row;
if (tsdbKeyCmprFn(&row, &pMemData->maxKey) > 0) pMemData->maxKey = *(TSDBKEY *)&row;
}
if (tsdbKeyCmprFn(&pMemTable->minKey, &pMemData->minKey) < 0) pMemTable->minKey = pMemData->minKey;
if (tsdbKeyCmprFn(&pMemTable->maxKey, &pMemData->maxKey) > 0) pMemTable->maxKey = pMemData->maxKey;
return code;
_err:
return code;
}
int32_t tsdbDeleteTableData2(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey) {
int32_t code = 0;
SMemTable *pMemTable = (SMemTable *)pTsdb->mem; // TODO
SMemData *pMemData;
SVBufPool *pPool = pTsdb->pVnode->inUse;
ASSERT(pMemTable);
{
// check if table exists (todo)
}
code = tsdbGetOrCreateMemData(pMemTable, suid, uid, &pMemData);
if (code) {
goto _err;
}
// do delete
SDelOp *pDelOp = (SDelOp *)vnodeBufPoolMalloc(pPool, sizeof(*pDelOp));
if (pDelOp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pDelOp->version = version;
pDelOp->sKey = sKey;
pDelOp->eKey = eKey;
pDelOp->pNext = NULL;
if (pMemData->delOpHead == NULL) {
ASSERT(pMemData->delOpTail == NULL);
pMemData->delOpHead = pMemData->delOpTail = pDelOp;
} else {
pMemData->delOpTail->pNext = pDelOp;
pMemData->delOpTail = pDelOp;
}
{
// update the state of pMemTable, pMemData, last and lastrow (todo)
}
tsdbDebug("vgId:%d delete data from table suid:%" PRId64 " uid:%" PRId64 " sKey:%" PRId64 " eKey:%" PRId64
" since %s",
TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
return code;
_err:
tsdbError("vgId:%d failed to delete data from table suid:%" PRId64 " uid:%" PRId64 " sKey:%" PRId64 " eKey:%" PRId64
" since %s",
TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
return code;
}
static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData) {
int32_t code = 0;
int32_t idx = 0;
SMemData *pMemDataT = &(SMemData){.suid = suid, .uid = uid};
SMemData *pMemData = NULL;
SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
int8_t maxLevel = pMemTable->pTsdb->pVnode->config.tsdbCfg.slLevel;
// get
idx = taosArraySearchIdx(pMemTable->pArray, &pMemDataT, memDataPCmprFn, TD_GE);
if (idx >= 0) {
pMemData = (SMemData *)taosArrayGet(pMemTable->pArray, idx);
if (memDataPCmprFn(&pMemDataT, &pMemData) == 0) goto _exit;
}
// create
pMemData = vnodeBufPoolMalloc(pPool, sizeof(*pMemData) + SL_NODE_HALF_SIZE(maxLevel) * 2);
if (pMemData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pMemData->suid = suid;
pMemData->uid = uid;
pMemData->minKey = (TSDBKEY){.version = INT64_MAX, .ts = TSKEY_MAX};
pMemData->maxKey = (TSDBKEY){.version = -1, .ts = TSKEY_MIN};
pMemData->delOpHead = pMemData->delOpTail = NULL;
pMemData->sl.seed = taosRand();
pMemData->sl.size = 0;
pMemData->sl.maxLevel = maxLevel;
pMemData->sl.level = 0;
pMemData->sl.pHead = (SMemSkipListNode *)&pMemData[1];
pMemData->sl.pTail = (SMemSkipListNode *)POINTER_SHIFT(pMemData->sl.pHead, SL_NODE_HALF_SIZE(maxLevel));
for (int8_t iLevel = 0; iLevel < pMemData->sl.maxLevel; iLevel++) {
SL_HEAD_FORWARD(&pMemData->sl, iLevel) = pMemData->sl.pTail;
SL_TAIL_BACKWARD(&pMemData->sl, iLevel) = pMemData->sl.pHead;
}
if (idx < 0) idx = 0;
if (taosArrayInsert(pMemTable->pArray, idx, &pMemData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
_exit:
*ppMemData = pMemData;
return code;
_err:
*ppMemData = NULL;
return code;
}
static int memDataPCmprFn(const void *p1, const void *p2) {
SMemData *pMemData1 = *(SMemData **)p1;
SMemData *pMemData2 = *(SMemData **)p2;
if (pMemData1->suid < pMemData2->suid) {
return -1;
} else if (pMemData1->suid > pMemData2->suid) {
return 1;
}
if (pMemData1->uid < pMemData2->uid) {
return -1;
} else if (pMemData1->uid > pMemData2->uid) {
return 1;
}
return 0;
}
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow) {
int32_t n = 0;
n += tPutI64(p ? p + n : p, pRow->version);
n += tPutTSRow(p ? p + n : p, &pRow->tsRow);
return n;
}
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow) {
int32_t n = 0;
n += tGetI64(p + n, &pRow->version);
n += tGetTSRow(p + n, &pRow->tsRow);
return n;
}
static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
int8_t level = 1;
int8_t tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
const uint32_t factor = 4;
while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
level++;
}
return level;
}
#if 0 //====================================================================================
#define SL_MAX_LEVEL 5
struct SMemSkipListCurosr {
SMemSkipList *pSl;
SMemSkipListNode *pNodes[SL_MAX_LEVEL];
@ -74,12 +330,6 @@ typedef struct {
#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
#define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level))
#define SL_HEAD_NODE(sl) ((sl)->pHead)
#define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel))
#define SL_HEAD_NODE_FORWARD(n, l) SL_NODE_FORWARD(n, l)
@ -99,50 +349,7 @@ static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc);
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow);
// SMemTable
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
SMemTable *pMemTb = NULL;
pMemTb = taosMemoryCalloc(1, sizeof(*pMemTb));
if (pMemTb == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMemTb->pTsdb = pTsdb;
pMemTb->minKey = TSKEY_MAX;
pMemTb->maxKey = TSKEY_MIN;
pMemTb->minVer = -1;
pMemTb->maxVer = -1;
pMemTb->nRows = 0;
pMemTb->nHash = 0;
pMemTb->nBucket = 1024;
pMemTb->pBuckets = taosMemoryCalloc(pMemTb->nBucket, sizeof(*pMemTb->pBuckets));
if (pMemTb->pBuckets == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pMemTb);
return -1;
}
if (tsdbMemSkipListCursorCreate(pTsdb->pVnode->config.tsdbCfg.slLevel, &pMemTb->pSlc) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pMemTb->pBuckets);
taosMemoryFree(pMemTb);
}
*ppMemTb = pMemTb;
return 0;
}
int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) {
if (pMemTb) {
// loop to destroy the contents (todo)
tsdbMemSkipListCursorDestroy(pMemTb->pSlc);
taosMemoryFree(pMemTb->pBuckets);
taosMemoryFree(pMemTb);
}
return 0;
}
// SMemTable ========================
int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) {
SMemData *pMemData;
STsdb *pTsdb = pMemTb->pTsdb;
@ -253,18 +460,6 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
return 0;
}
static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
int8_t level = 1;
int8_t tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
const uint32_t factor = 4;
while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
level++;
}
return level;
}
static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow) {
if (tEncodeI64(pEncoder, pRow->version) < 0) return -1;
if (tEncodeBinary(pEncoder, (const uint8_t *)pRow->pRow, pRow->szRow) < 0) return -1;
@ -377,4 +572,5 @@ static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipLis
}
return pNode;
}
}
#endif

2203
source/dnode/vnode/src/tsdb/tsdbSma.c
View File

File diff suppressed because it is too large Load Diff

128
source/dnode/vnode/src/tsdb/tsdbTDBImpl.c
View File

@ -1,128 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#include "tsdb.h"
int32_t tsdbOpenDBEnv(TDB **ppEnv, const char *path) {
int ret = 0;
if (path == NULL) return -1;
ret = tdbOpen(path, 4096, 256, ppEnv); // use as param
if (ret != 0) {
tsdbError("Failed to create tsdb db env, ret = %d", ret);
return -1;
}
return 0;
}
int32_t tsdbCloseDBEnv(TDB *pEnv) { return tdbClose(pEnv); }
static inline int tsdbSmaKeyCmpr(const void *arg1, int len1, const void *arg2, int len2) {
const SSmaKey *pKey1 = (const SSmaKey *)arg1;
const SSmaKey *pKey2 = (const SSmaKey *)arg2;
ASSERT(len1 == len2 && len1 == sizeof(SSmaKey));
if (pKey1->skey < pKey2->skey) {
return -1;
} else if (pKey1->skey > pKey2->skey) {
return 1;
}
if (pKey1->groupId < pKey2->groupId) {
return -1;
} else if (pKey1->groupId > pKey2->groupId) {
return 1;
}
return 0;
}
static int32_t tsdbOpenDBDb(TTB **ppDB, TDB *pEnv, const char *pFName) {
int ret;
tdb_cmpr_fn_t compFunc;
// Create a database
compFunc = tsdbSmaKeyCmpr;
ret = tdbTbOpen(pFName, -1, -1, compFunc, pEnv, ppDB);
return 0;
}
static int32_t tsdbCloseDBDb(TTB *pDB) { return tdbTbClose(pDB); }
int32_t tsdbOpenDBF(TDB *pEnv, SDBFile *pDBF) {
// TEnv is shared by a group of SDBFile
if (!pEnv || !pDBF) {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
// Open DBF
if (tsdbOpenDBDb(&(pDBF->pDB), pEnv, pDBF->path) < 0) {
terrno = TSDB_CODE_TDB_INIT_FAILED;
tsdbCloseDBDb(pDBF->pDB);
return -1;
}
return 0;
}
int32_t tsdbCloseDBF(SDBFile *pDBF) {
int32_t ret = 0;
if (pDBF->pDB) {
ret = tsdbCloseDBDb(pDBF->pDB);
pDBF->pDB = NULL;
}
taosMemoryFreeClear(pDBF->path);
return ret;
}
int32_t tsdbSaveSmaToDB(SDBFile *pDBF, void *pKey, int32_t keyLen, void *pVal, int32_t valLen, TXN *txn) {
int32_t ret;
ret = tdbTbInsert(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
if (ret < 0) {
tsdbError("Failed to create insert sma data into db, ret = %d", ret);
return -1;
}
return 0;
}
void *tsdbGetSmaDataByKey(SDBFile *pDBF, const void *pKey, int32_t keyLen, int32_t *valLen) {
void *pVal = NULL;
int ret;
ret = tdbTbGet(pDBF->pDB, pKey, keyLen, &pVal, valLen);
if (ret < 0) {
tsdbError("Failed to get sma data from db, ret = %d", ret);
return NULL;
}
ASSERT(*valLen >= 0);
// TODO: lock?
// TODO: Would the key/value be destoryed during return the data?
// TODO: How about the key is updated while value length is changed? The original value buffer would be freed
// automatically?
return pVal;
}

86
source/libs/parser/src/parInsert.c
View File

@ -780,8 +780,8 @@ static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTa
return;
}
static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema,
int16_t timePrec, STagVal *val, SMsgBuf* pMsgBuf) {
static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16_t timePrec, STagVal* val,
SMsgBuf* pMsgBuf) {
int64_t iv;
uint64_t uv;
char* endptr = NULL;
@ -937,8 +937,8 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema,
case TSDB_DATA_TYPE_NCHAR: {
int32_t output = 0;
void *p = taosMemoryCalloc(1, pToken->n * TSDB_NCHAR_SIZE);
if(p == NULL){
void* p = taosMemoryCalloc(1, pToken->n * TSDB_NCHAR_SIZE);
if (p == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (!taosMbsToUcs4(pToken->z, pToken->n, (TdUcs4*)(p), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
@ -971,11 +971,11 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema,
// pSql -> tag1_value, ...)
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) {
int32_t code = TSDB_CODE_SUCCESS;
SArray *pTagVals = taosArrayInit(pCxt->tags.numOfBound, sizeof(STagVal));
SToken sToken;
bool isParseBindParam = false;
bool isJson = false;
STag* pTag = NULL;
SArray* pTagVals = taosArrayInit(pCxt->tags.numOfBound, sizeof(STagVal));
SToken sToken;
bool isParseBindParam = false;
bool isJson = false;
STag* pTag = NULL;
for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
@ -995,13 +995,13 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
}
SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]];
char *tmpTokenBuf = taosMemoryCalloc(1, sToken.n); // this can be optimize with parse column
char* tmpTokenBuf = taosMemoryCalloc(1, sToken.n); // this can be optimize with parse column
code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg);
if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(tmpTokenBuf);
goto end;
}
if(pTagSchema->type == TSDB_DATA_TYPE_JSON){
if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
if (sToken.n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
code = buildSyntaxErrMsg(&pCxt->msg, "json string too long than 4095", sToken.z);
taosMemoryFree(tmpTokenBuf);
@ -1009,18 +1009,18 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
}
code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
taosMemoryFree(tmpTokenBuf);
if(code != TSDB_CODE_SUCCESS){
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
isJson = true;
}else{
} else {
STagVal val = {0};
code = parseTagToken(&pCxt->pSql, &sToken, pTagSchema, precision, &val, &pCxt->msg);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(tmpTokenBuf);
goto end;
}
if (pTagSchema->type != TSDB_DATA_TYPE_BINARY){
if (pTagSchema->type != TSDB_DATA_TYPE_BINARY) {
taosMemoryFree(tmpTokenBuf);
}
taosArrayPush(pTagVals, &val);
@ -1032,7 +1032,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
goto end;
}
if(!isJson && (code = tTagNew(pTagVals, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
if (!isJson && (code = tTagNew(pTagVals, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
goto end;
}
@ -1040,8 +1040,8 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
end:
for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
STagVal *p = (STagVal *)taosArrayGet(pTagVals, i);
if(IS_VAR_DATA_TYPE(p->type)){
STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
if (IS_VAR_DATA_TYPE(p->type)) {
taosMemoryFree(p->pData);
}
}
@ -1701,10 +1701,10 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
return buildInvalidOperationMsg(&pBuf, "out of memory");
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t code = TSDB_CODE_SUCCESS;
SSchema* pSchema = pDataBlock->pTableMeta->schema;
bool isJson = false;
bool isJson = false;
STag* pTag = NULL;
for (int c = 0; c < tags->numOfBound; ++c) {
@ -1713,7 +1713,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
}
SSchema* pTagSchema = &pSchema[tags->boundColumns[c]];
int32_t colLen = pTagSchema->bytes;
int32_t colLen = pTagSchema->bytes;
if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
colLen = bind[c].length[0];
}
@ -1724,22 +1724,22 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
}
isJson = true;
char *tmp = taosMemoryCalloc(1, colLen + 1);
char* tmp = taosMemoryCalloc(1, colLen + 1);
memcpy(tmp, bind[c].buffer, colLen);
code = parseJsontoTagData(tmp, pTagArray, &pTag, &pBuf);
taosMemoryFree(tmp);
if(code != TSDB_CODE_SUCCESS){
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
}else{
} else {
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
if(pTagSchema->type == TSDB_DATA_TYPE_BINARY){
if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
val.pData = (uint8_t*)bind[c].buffer;
val.nData = colLen;
}else if(pTagSchema->type == TSDB_DATA_TYPE_NCHAR){
} else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
int32_t output = 0;
void *p = taosMemoryCalloc(1, colLen * TSDB_NCHAR_SIZE);
if(p == NULL){
void* p = taosMemoryCalloc(1, colLen * TSDB_NCHAR_SIZE);
if (p == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
@ -1757,7 +1757,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
}
val.pData = p;
val.nData = output;
}else{
} else {
memcpy(&val.i64, bind[c].buffer, colLen);
}
taosArrayPush(pTagArray, &val);
@ -1775,8 +1775,8 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
end:
for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
STagVal *p = (STagVal *)taosArrayGet(pTagArray, i);
if(p->type == TSDB_DATA_TYPE_NCHAR){
STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
if (p->type == TSDB_DATA_TYPE_NCHAR) {
taosMemoryFree(p->pData);
}
}
@ -1951,7 +1951,8 @@ int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBu
return TSDB_CODE_SUCCESS;
}
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_E** fields, uint8_t timePrec) {
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_E** fields,
uint8_t timePrec) {
if (fields) {
*fields = taosMemoryCalloc(boundInfo->numOfBound, sizeof(TAOS_FIELD));
if (NULL == *fields) {
@ -1962,7 +1963,7 @@ int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_
if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
(*fields)[0].precision = timePrec;
}
for (int32_t i = 0; i < boundInfo->numOfBound; ++i) {
schema = &pSchema[boundInfo->boundColumns[i]];
strcpy((*fields)[i].name, schema->name);
@ -2008,7 +2009,8 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_E** fiel
return TSDB_CODE_SUCCESS;
}
CHECK_CODE(buildBoundFields(&pDataBlock->boundColumnInfo, pSchema, fieldNum, fields, pDataBlock->pTableMeta->tableInfo.precision));
CHECK_CODE(buildBoundFields(&pDataBlock->boundColumnInfo, pSchema, fieldNum, fields,
pDataBlock->pTableMeta->tableInfo.precision));
return TSDB_CODE_SUCCESS;
}
@ -2122,16 +2124,16 @@ static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* p
int32_t code = TSDB_CODE_SUCCESS;
for (int i = 0; i < tags->numOfBound; ++i) {
SSchema* pTagSchema = &pSchema[tags->boundColumns[i]];
SSmlKv* kv = taosArrayGetP(cols, i);
SSmlKv* kv = taosArrayGetP(cols, i);
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
if(pTagSchema->type == TSDB_DATA_TYPE_BINARY){
val.pData = (uint8_t *)kv->value;
if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
val.pData = (uint8_t*)kv->value;
val.nData = kv->length;
}else if(pTagSchema->type == TSDB_DATA_TYPE_NCHAR){
} else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
int32_t output = 0;
void *p = taosMemoryCalloc(1, pTagSchema->bytes - VARSTR_HEADER_SIZE);
if(p == NULL){
void* p = taosMemoryCalloc(1, pTagSchema->bytes - VARSTR_HEADER_SIZE);
if (p == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
@ -2149,7 +2151,7 @@ static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* p
}
val.pData = p;
val.nData = output;
}else{
} else {
memcpy(&val.i64, &(kv->value), kv->length);
}
taosArrayPush(pTagArray, &val);
@ -2158,8 +2160,8 @@ static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* p
code = tTagNew(pTagArray, 1, false, ppTag);
end:
for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
STagVal *p = (STagVal *)taosArrayGet(pTagArray, i);
if(p->type == TSDB_DATA_TYPE_NCHAR){
STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
if (p->type == TSDB_DATA_TYPE_NCHAR) {
taosMemoryFree(p->pData);
}
}