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

Merge pull request #25707 from taosdata/fix/TD-29981 

fix: refactor lastcache ser/des for compatibility
This commit is contained in:
Hongze Cheng 2024-05-10 09:15:29 +08:00 committed by GitHub
parent ba0d0a316f 0ac5d393e6
commit db68b9a1f0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 157 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/executor/storageapi.h
View File

@ -35,7 +35,6 @@ extern "C" {
#define CACHESCAN_RETRIEVE_TYPE_SINGLE 0x2 #define CACHESCAN_RETRIEVE_TYPE_SINGLE 0x2
#define CACHESCAN_RETRIEVE_LAST_ROW 0x4 #define CACHESCAN_RETRIEVE_LAST_ROW 0x4
#define CACHESCAN_RETRIEVE_LAST 0x8 #define CACHESCAN_RETRIEVE_LAST 0x8
#define CACHESCAN_RETRIEVE_PK 0x10
#define META_READER_LOCK 0x0 #define META_READER_LOCK 0x0
#define META_READER_NOLOCK 0x1 #define META_READER_NOLOCK 0x1

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

@ -894,35 +894,14 @@ typedef enum {
READER_EXEC_ROWS = 0x2, READER_EXEC_ROWS = 0x2,
} EExecMode; } EExecMode;
#define LAST_COL_VERSION (0x1)
typedef struct { typedef struct {
SRowKey rowKey; SRowKey rowKey;
int8_t dirty; int8_t dirty;
SColVal colVal; SColVal colVal;
} SLastCol; } SLastCol;
typedef struct {
union {
int64_t val;
struct {
uint8_t *pData;
uint32_t nData;
};
};
} SValueV0;
typedef struct {
int16_t cid;
int8_t type;
int8_t flag;
SValueV0 value;
} SColValV0;
typedef struct {
TSKEY ts;
int8_t dirty;
SColValV0 colVal;
} SLastColV0;
int32_t tsdbOpenCache(STsdb *pTsdb); int32_t tsdbOpenCache(STsdb *pTsdb);
void tsdbCloseCache(STsdb *pTsdb); void tsdbCloseCache(STsdb *pTsdb);
int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *row); int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *row);

314
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -130,20 +130,14 @@ static void tsdbClosePgCache(STsdb *pTsdb) {
enum { enum {
LFLAG_LAST_ROW = 0, LFLAG_LAST_ROW = 0,
LFLAG_LAST = 1, LFLAG_LAST = 1,
LFLAG_VERSION = 1 << 2,
LFLAG_VERSION_BITS = (1 << 2 | 1 << 3),
LFLAG_PRIMARY_KEY = CACHESCAN_RETRIEVE_PK,
}; };
#define LAST_KEY_HAS_VERSION ((k).lflag & LFLAG_VERSION_BITS)
typedef struct { typedef struct {
tb_uid_t uid; tb_uid_t uid;
int16_t cid; int16_t cid;
int8_t lflag; int8_t lflag;
} SLastKey; } SLastKey;
#define HAS_PRIMARY_KEY(k) (((k).lflag & LFLAG_PRIMARY_KEY) == LFLAG_PRIMARY_KEY)
#define IS_LAST_ROW_KEY(k) (((k).lflag & LFLAG_LAST) == LFLAG_LAST_ROW) #define IS_LAST_ROW_KEY(k) (((k).lflag & LFLAG_LAST) == LFLAG_LAST_ROW)
#define IS_LAST_KEY(k) (((k).lflag & LFLAG_LAST) == LFLAG_LAST) #define IS_LAST_KEY(k) (((k).lflag & LFLAG_LAST) == LFLAG_LAST)
@ -336,142 +330,163 @@ static void rocksMayWrite(STsdb *pTsdb, bool force, bool read, bool lock) {
} }
} }
// note: new object do not own colVal's resource, just copy the pointer typedef struct {
static SLastCol *tsdbCacheConvertLastColV0(SLastColV0 *pLastColV0) { TSKEY ts;
SLastCol *pLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); int8_t dirty;
if (pLastCol == NULL) return NULL; struct {
int16_t cid;
int8_t type;
int8_t flag;
union {
int64_t val;
struct {
uint32_t nData;
uint8_t *pData;
};
} value;
} colVal;
} SLastColV0;
static int32_t tsdbCacheDeserializeV0(char const *value, SLastCol *pLastCol) {
SLastColV0 *pLastColV0 = (SLastColV0 *)value;
pLastCol->rowKey.ts = pLastColV0->ts; pLastCol->rowKey.ts = pLastColV0->ts;
pLastCol->rowKey.numOfPKs = 0; pLastCol->rowKey.numOfPKs = 0;
pLastCol->dirty = pLastColV0->dirty; pLastCol->dirty = pLastColV0->dirty;
pLastCol->colVal.cid = pLastColV0->colVal.cid; pLastCol->colVal.cid = pLastColV0->colVal.cid;
pLastCol->colVal.flag = pLastColV0->colVal.flag; pLastCol->colVal.flag = pLastColV0->colVal.flag;
pLastCol->colVal.value.type = pLastColV0->colVal.type; pLastCol->colVal.value.type = pLastColV0->colVal.type;
pLastCol->colVal.value.val = pLastColV0->colVal.value.val;
return pLastCol; if (IS_VAR_DATA_TYPE(pLastCol->colVal.value.type)) {
pLastCol->colVal.value.nData = pLastColV0->colVal.value.nData;
pLastCol->colVal.value.pData = (uint8_t *)(&pLastColV0[1]);
return sizeof(SLastColV0) + pLastColV0->colVal.value.nData;
} else {
pLastCol->colVal.value.val = pLastColV0->colVal.value.val;
return sizeof(SLastColV0);
}
} }
static SLastCol *tsdbCacheDeserializeV0(char const *value) { static SLastCol *tsdbCacheDeserialize(char const *value, size_t size) {
if (!value) { if (!value) {
return NULL; return NULL;
} }
SLastCol* pLastCol = taosMemoryCalloc(1, sizeof(SLastCol));
if (NULL == pLastCol) {
return NULL;
}
int32_t offset = tsdbCacheDeserializeV0(value, pLastCol);
if (offset == size) {
// version 0
return pLastCol;
} else if (offset > size) {
terrno = TSDB_CODE_INVALID_DATA_FMT;
taosMemoryFreeClear(pLastCol);
return NULL;
}
// version
int8_t version = *(int8_t *)(value + offset);
offset += sizeof(int8_t);
// numOfPKs
pLastCol->rowKey.numOfPKs = *(uint8_t *)(value + offset);
offset += sizeof(uint8_t);
// pks
for (int32_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) {
pLastCol->rowKey.pks[i] = *(SValue *)(value + offset);
offset += sizeof(SValue);
if (IS_VAR_DATA_TYPE(pLastCol->rowKey.pks[i].type)) {
pLastCol->rowKey.pks[i].pData = (uint8_t *)value + offset;
offset += pLastCol->rowKey.pks[i].nData;
}
}
if (offset > size) {
terrno = TSDB_CODE_INVALID_DATA_FMT;
taosMemoryFreeClear(pLastCol);
return NULL;
}
return pLastCol;
}
/*
typedef struct {
SLastColV0 lastColV0;
char colData[];
int8_t version;
uint8_t numOfPKs;
SValue pks[0];
char pk0Data[];
SValue pks[1];
char pk1Data[];
...
} SLastColDisk;
*/
static int32_t tsdbCacheSerializeV0(char const *value, SLastCol *pLastCol) {
SLastColV0 *pLastColV0 = (SLastColV0 *)value; SLastColV0 *pLastColV0 = (SLastColV0 *)value;
SColValV0 *pColVal = &pLastColV0->colVal;
if (IS_VAR_DATA_TYPE(pColVal->type)) { pLastColV0->ts = pLastCol->rowKey.ts;
if (pColVal->value.nData > 0) { pLastColV0->dirty = pLastCol->dirty;
pColVal->value.pData = (char *)value + sizeof(*pLastColV0); pLastColV0->colVal.cid = pLastCol->colVal.cid;
} else { pLastColV0->colVal.flag = pLastCol->colVal.flag;
pColVal->value.pData = NULL; pLastColV0->colVal.type = pLastCol->colVal.value.type;
if (IS_VAR_DATA_TYPE(pLastCol->colVal.value.type)) {
pLastColV0->colVal.value.nData = pLastCol->colVal.value.nData;
if (pLastCol->colVal.value.nData > 0) {
memcpy(&pLastColV0[1], pLastCol->colVal.value.pData, pLastCol->colVal.value.nData);
} }
return sizeof(SLastColV0) + pLastCol->colVal.value.nData;
} else {
pLastColV0->colVal.value.val = pLastCol->colVal.value.val;
return sizeof(SLastColV0);
} }
return tsdbCacheConvertLastColV0(pLastColV0); return 0;
}
static SLastCol *tsdbCacheDeserializeV1(char const *value) {
if (!value) {
return NULL;
}
SLastCol *pLastCol = taosMemoryMalloc(sizeof(SLastCol));
*pLastCol = *(SLastCol *)(value);
char *currentPos = (char *)value + sizeof(*pLastCol);
for (int8_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) {
SValue *pValue = &pLastCol->rowKey.pks[i];
if (IS_VAR_DATA_TYPE(pValue->type)) {
if (pValue->nData > 0) {
pValue->pData = currentPos;
currentPos += pValue->nData;
} else {
pValue->pData = NULL;
}
}
}
SColVal *pColVal = &pLastCol->colVal;
if (IS_VAR_DATA_TYPE(pColVal->value.type)) {
if (pColVal->value.nData > 0) {
pColVal->value.pData = currentPos;
currentPos += pColVal->value.nData;
} else {
pColVal->value.pData = NULL;
}
}
return pLastCol;
}
static SLastCol *tsdbCacheDeserialize(char const *value, int8_t lflag) {
if (!value) {
return NULL;
}
int8_t version = lflag & LFLAG_VERSION_BITS;
SLastCol *lastCol = NULL;
switch (version) {
case 0:
lastCol = tsdbCacheDeserializeV0(value);
break;
case LFLAG_VERSION:
lastCol = tsdbCacheDeserializeV1(value);
break;
defalut:
tsdbError("invalid last key version %" PRId8 " , lflag:%" PRId8, version, lflag);
break;
}
return lastCol;
}
static uint32_t tsdbCacheCopyVarData(SValue *from, SValue *to) {
ASSERT(from->nData >= 0);
if (from->nData > 0) {
memcpy(to->pData, from->pData, from->nData);
}
to->type = from->type;
to->nData = from->nData;
return from->nData;
} }
static void tsdbCacheSerialize(SLastCol *pLastCol, char **value, size_t *size) { static void tsdbCacheSerialize(SLastCol *pLastCol, char **value, size_t *size) {
SColVal *pColVal = &pLastCol->colVal; *size = sizeof(SLastColV0);
size_t length = sizeof(*pLastCol); if (IS_VAR_DATA_TYPE(pLastCol->colVal.value.type)) {
*size += pLastCol->colVal.value.nData;
}
*size += sizeof(uint8_t) + sizeof(uint8_t); // version + numOfPKs
for (int8_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) { for (int8_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) {
*size += sizeof(SValue);
if (IS_VAR_DATA_TYPE(pLastCol->rowKey.pks[i].type)) { if (IS_VAR_DATA_TYPE(pLastCol->rowKey.pks[i].type)) {
length += pLastCol->rowKey.pks[i].nData; *size += pLastCol->rowKey.pks[i].nData;
} }
} }
if (IS_VAR_DATA_TYPE(pColVal->value.type)) {
length += pColVal->value.nData;
}
*value = taosMemoryMalloc(length); *value = taosMemoryMalloc(*size);
// copy last col int32_t offset = tsdbCacheSerializeV0(*value, pLastCol);
SLastCol *pToLastCol = (SLastCol *)(*value);
*pToLastCol = *pLastCol;
char *currentPos = *value + sizeof(*pLastCol);
// copy var data pks // version
((uint8_t *)(*value + offset))[0] = LAST_COL_VERSION;
offset++;
// numOfPKs
((uint8_t *)(*value + offset))[0] = pLastCol->rowKey.numOfPKs;
offset++;
// pks
for (int8_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) { for (int8_t i = 0; i < pLastCol->rowKey.numOfPKs; i++) {
SValue *pFromValue = &pLastCol->rowKey.pks[i]; ((SValue *)(*value + offset))[0] = pLastCol->rowKey.pks[i];
if (IS_VAR_DATA_TYPE(pFromValue->type)) { offset += sizeof(SValue);
SValue *pToValue = &pToLastCol->rowKey.pks[i]; if (IS_VAR_DATA_TYPE(pLastCol->rowKey.pks[i].type)) {
pToValue->pData = (pFromValue->nData == 0) ? NULL : currentPos; if (pLastCol->rowKey.pks[i].nData > 0) {
currentPos += tsdbCacheCopyVarData(pFromValue, pToValue); memcpy(*value + offset, pLastCol->rowKey.pks[i].pData, pLastCol->rowKey.pks[i].nData);
}
offset += pLastCol->rowKey.pks[i].nData;
}
} }
}
// copy var data value
if (IS_VAR_DATA_TYPE(pColVal->value.type)) {
SValue *pFromValue = &pColVal->value;
SValue *pToValue = &pToLastCol->colVal.value;
pToValue->pData = (pFromValue->nData == 0) ? NULL : currentPos;
currentPos += tsdbCacheCopyVarData(pFromValue, pToValue);
}
*size = length;
} }
static void tsdbCachePutBatch(SLastCol *pLastCol, const void *key, size_t klen, SCacheFlushState *state) { static void tsdbCachePutBatch(SLastCol *pLastCol, const void *key, size_t klen, SCacheFlushState *state) {
@ -603,7 +618,7 @@ static int32_t tsdbCacheNewTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, i
charge += pLastCol->colVal.value.nData; charge += pLastCol->colVal.value.nData;
} }
SLastKey *pLastKey = &(SLastKey){.lflag = lflag | LFLAG_VERSION, .uid = uid, .cid = cid}; SLastKey *pLastKey = &(SLastKey){.lflag = lflag, .uid = uid, .cid = cid};
LRUStatus status = taosLRUCacheInsert(pCache, pLastKey, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter, NULL, LRUStatus status = taosLRUCacheInsert(pCache, pLastKey, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter, NULL,
TAOS_LRU_PRIORITY_LOW, &pTsdb->flushState); TAOS_LRU_PRIORITY_LOW, &pTsdb->flushState);
if (status != TAOS_LRU_STATUS_OK) { if (status != TAOS_LRU_STATUS_OK) {
@ -653,11 +668,9 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
size_t *keys_list_sizes = taosMemoryCalloc(2, sizeof(size_t)); size_t *keys_list_sizes = taosMemoryCalloc(2, sizeof(size_t));
const size_t klen = ROCKS_KEY_LEN; const size_t klen = ROCKS_KEY_LEN;
int8_t lflag = hasPrimaryKey ? LFLAG_PRIMARY_KEY : 0;
char *keys = taosMemoryCalloc(2, sizeof(SLastKey)); char *keys = taosMemoryCalloc(2, sizeof(SLastKey));
((SLastKey *)keys)[0] = (SLastKey){.lflag = lflag | LFLAG_LAST | LFLAG_VERSION, .uid = uid, .cid = cid}; ((SLastKey *)keys)[0] = (SLastKey){.lflag = LFLAG_LAST, .uid = uid, .cid = cid};
((SLastKey *)keys)[1] = (SLastKey){.lflag = lflag | LFLAG_LAST_ROW | LFLAG_VERSION, .uid = uid, .cid = cid}; ((SLastKey *)keys)[1] = (SLastKey){.lflag = LFLAG_LAST_ROW, .uid = uid, .cid = cid};
keys_list[0] = keys; keys_list[0] = keys;
keys_list[1] = keys + sizeof(SLastKey); keys_list[1] = keys + sizeof(SLastKey);
@ -681,13 +694,13 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
{ {
SLastCol *pLastCol = tsdbCacheDeserialize(values_list[0], ((SLastKey*)keys_list[0])->lflag); SLastCol *pLastCol = tsdbCacheDeserialize(values_list[0], values_list_sizes[0]);
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen); rocksdb_writebatch_delete(wb, keys_list[0], klen);
} }
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
pLastCol = tsdbCacheDeserialize(values_list[1], ((SLastKey*)keys_list[1])->lflag); pLastCol = tsdbCacheDeserialize(values_list[1], values_list_sizes[1]);
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen); rocksdb_writebatch_delete(wb, keys_list[1], klen);
} }
@ -733,18 +746,12 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
taosThreadMutexLock(&pTsdb->lruMutex); taosThreadMutexLock(&pTsdb->lruMutex);
if (suid < 0) { if (suid < 0) {
int8_t lflag = 0; for (int i = 0; i < pSchemaRow->nCols; ++i) {
int nCols = pSchemaRow->nCols;
if (nCols >= 2) {
lflag = (pSchemaRow->pSchema[1].flags & COL_IS_KEY) ? LFLAG_PRIMARY_KEY : 0;
}
for (int i = 0; i < nCols; ++i) {
int16_t cid = pSchemaRow->pSchema[i].colId; int16_t cid = pSchemaRow->pSchema[i].colId;
int8_t col_type = pSchemaRow->pSchema[i].type; int8_t col_type = pSchemaRow->pSchema[i].type;
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, lflag | LFLAG_LAST_ROW); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, lflag | LFLAG_LAST); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
} }
} else { } else {
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
@ -754,18 +761,12 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
return -1; return -1;
} }
int8_t lflag = 0; for (int i = 0; i < pTSchema->numOfCols; ++i) {
int nCols = pTSchema->numOfCols;
if (nCols >= 2) {
lflag = (pTSchema->columns[1].flags & COL_IS_KEY) ? LFLAG_PRIMARY_KEY : 0;
}
for (int i = 0; i < nCols; ++i) {
int16_t cid = pTSchema->columns[i].colId; int16_t cid = pTSchema->columns[i].colId;
int8_t col_type = pTSchema->columns[i].type; int8_t col_type = pTSchema->columns[i].type;
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, lflag | LFLAG_LAST_ROW); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, lflag | LFLAG_LAST); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
} }
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
@ -1024,14 +1025,13 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
STsdbRowKey tsdbRowKey = {0}; STsdbRowKey tsdbRowKey = {0};
tsdbRowGetKey(pRow, &tsdbRowKey); tsdbRowGetKey(pRow, &tsdbRowKey);
SRowKey *pRowKey = &tsdbRowKey.key; SRowKey *pRowKey = &tsdbRowKey.key;
int8_t lflag = (pRowKey->numOfPKs != 0) ? LFLAG_PRIMARY_KEY : 0;
taosThreadMutexLock(&pTsdb->lruMutex); taosThreadMutexLock(&pTsdb->lruMutex);
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
SColVal *pColVal = (SColVal *)taosArrayGet(aColVal, i); SColVal *pColVal = (SColVal *)taosArrayGet(aColVal, i);
int16_t cid = pColVal->cid; int16_t cid = pColVal->cid;
SLastKey *key = &(SLastKey){.lflag = lflag | LFLAG_LAST_ROW | LFLAG_VERSION, .uid = uid, .cid = cid}; SLastKey *key = &(SLastKey){.lflag = LFLAG_LAST_ROW, .uid = uid, .cid = cid};
size_t klen = ROCKS_KEY_LEN; size_t klen = ROCKS_KEY_LEN;
LRUHandle *h = taosLRUCacheLookup(pCache, key, klen); LRUHandle *h = taosLRUCacheLookup(pCache, key, klen);
if (h) { if (h) {
@ -1049,7 +1049,7 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
} }
if (COL_VAL_IS_VALUE(pColVal)) { if (COL_VAL_IS_VALUE(pColVal)) {
key->lflag = lflag | LFLAG_LAST | LFLAG_VERSION; key->lflag = LFLAG_LAST;
LRUHandle *h = taosLRUCacheLookup(pCache, key, klen); LRUHandle *h = taosLRUCacheLookup(pCache, key, klen);
if (h) { if (h) {
SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pCache, h); SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pCache, h);
@ -1094,7 +1094,7 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
SColVal *pColVal = (SColVal *)TARRAY_DATA(aColVal) + idxKey->idx; SColVal *pColVal = (SColVal *)TARRAY_DATA(aColVal) + idxKey->idx;
// SColVal *pColVal = (SColVal *)taosArrayGet(aColVal, idxKey->idx); // SColVal *pColVal = (SColVal *)taosArrayGet(aColVal, idxKey->idx);
SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], ((SLastKey*)keys_list[i])->lflag); SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], values_list_sizes[i]);
SLastCol *PToFree = pLastCol; SLastCol *PToFree = pLastCol;
if (IS_LAST_ROW_KEY(idxKey->key)) { if (IS_LAST_ROW_KEY(idxKey->key)) {
@ -1106,8 +1106,8 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
if (NULL == pLastCol || cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) { if (NULL == pLastCol || cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) {
char *value = NULL; char *value = NULL;
size_t vlen = 0; size_t vlen = 0;
tsdbCacheSerialize(&(SLastCol){.rowKey = *pRowKey, .colVal = *pColVal}, &value, SLastCol lastColTmp = {.rowKey = *pRowKey, .colVal = *pColVal};
&vlen); tsdbCacheSerialize(&lastColTmp, &value, &vlen);
taosThreadMutexLock(&pTsdb->rCache.rMutex); taosThreadMutexLock(&pTsdb->rCache.rMutex);
@ -1115,7 +1115,7 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
pLastCol = (SLastCol *)value; pLastCol = &lastColTmp;
SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol));
*pTmpLastCol = *pLastCol; *pTmpLastCol = *pLastCol;
pLastCol = pTmpLastCol; pLastCol = pTmpLastCol;
@ -1147,7 +1147,8 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
if (NULL == pLastCol || (tRowKeyCompare(&pLastCol->rowKey, pRowKey) != 1)) { if (NULL == pLastCol || (tRowKeyCompare(&pLastCol->rowKey, pRowKey) != 1)) {
char *value = NULL; char *value = NULL;
size_t vlen = 0; size_t vlen = 0;
tsdbCacheSerialize(&(SLastCol){.rowKey = *pRowKey, .colVal = *pColVal}, &value, &vlen); SLastCol lastColTmp = {.rowKey = *pRowKey, .colVal = *pColVal};
tsdbCacheSerialize(&lastColTmp, &value, &vlen);
taosThreadMutexLock(&pTsdb->rCache.rMutex); taosThreadMutexLock(&pTsdb->rCache.rMutex);
@ -1155,7 +1156,7 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
pLastCol = (SLastCol *)value; pLastCol = &lastColTmp;
SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol));
*pTmpLastCol = *pLastCol; *pTmpLastCol = *pLastCol;
pLastCol = pTmpLastCol; pLastCol = pTmpLastCol;
@ -1401,7 +1402,7 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
SIdxKey *idxKey = taosArrayGet(remainCols, 0); SIdxKey *idxKey = taosArrayGet(remainCols, 0);
if (idxKey->key.cid != PRIMARYKEY_TIMESTAMP_COL_ID) { if (idxKey->key.cid != PRIMARYKEY_TIMESTAMP_COL_ID) {
SLastKey *key = &(SLastKey){.lflag = ltype | LFLAG_VERSION, .uid = uid, .cid = PRIMARYKEY_TIMESTAMP_COL_ID}; SLastKey *key = &(SLastKey){.lflag = ltype, .uid = uid, .cid = PRIMARYKEY_TIMESTAMP_COL_ID};
taosArrayInsert(remainCols, 0, &(SIdxKey){0, *key}); taosArrayInsert(remainCols, 0, &(SIdxKey){0, *key});
} }
@ -1570,7 +1571,7 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
SLRUCache *pCache = pTsdb->lruCache; SLRUCache *pCache = pTsdb->lruCache;
for (int i = 0, j = 0; i < num_keys && j < TARRAY_SIZE(remainCols); ++i) { for (int i = 0, j = 0; i < num_keys && j < TARRAY_SIZE(remainCols); ++i) {
SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], ((SLastKey*)keys_list[i])->lflag); SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], values_list_sizes[i]);
SLastCol *PToFree = pLastCol; SLastCol *PToFree = pLastCol;
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j];
if (pLastCol) { if (pLastCol) {
@ -1637,7 +1638,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
int16_t cid = ((int16_t *)TARRAY_DATA(pCidList))[i]; int16_t cid = ((int16_t *)TARRAY_DATA(pCidList))[i];
SLastKey key = {.lflag = ltype | LFLAG_VERSION, .uid = uid, .cid = cid}; SLastKey key = {.lflag = ltype, .uid = uid, .cid = cid};
// for select last_row, last case // for select last_row, last case
int32_t funcType = FUNCTION_TYPE_CACHE_LAST; int32_t funcType = FUNCTION_TYPE_CACHE_LAST;
if (pr->pFuncTypeList != NULL && taosArrayGetSize(pr->pFuncTypeList) > i) { if (pr->pFuncTypeList != NULL && taosArrayGetSize(pr->pFuncTypeList) > i) {
@ -1726,17 +1727,12 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
size_t *keys_list_sizes = taosMemoryCalloc(num_keys * 2, sizeof(size_t)); size_t *keys_list_sizes = taosMemoryCalloc(num_keys * 2, sizeof(size_t));
const size_t klen = ROCKS_KEY_LEN; const size_t klen = ROCKS_KEY_LEN;
int8_t lflag = 0;
if (num_keys >= 2) {
lflag = (pTSchema->columns[1].flags & COL_IS_KEY) ? LFLAG_PRIMARY_KEY : 0;
}
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
int16_t cid = pTSchema->columns[i].colId; int16_t cid = pTSchema->columns[i].colId;
char *keys = taosMemoryCalloc(2, sizeof(SLastKey)); char *keys = taosMemoryCalloc(2, sizeof(SLastKey));
((SLastKey *)keys)[0] = (SLastKey){.lflag = lflag | LFLAG_LAST | LFLAG_VERSION, .uid = uid, .cid = cid}; ((SLastKey *)keys)[0] = (SLastKey){.lflag = LFLAG_LAST, .uid = uid, .cid = cid};
((SLastKey *)keys)[1] = (SLastKey){.lflag = lflag | LFLAG_LAST_ROW | LFLAG_VERSION, .uid = uid, .cid = cid}; ((SLastKey *)keys)[1] = (SLastKey){.lflag = LFLAG_LAST_ROW, .uid = uid, .cid = cid};
keys_list[i] = keys; keys_list[i] = keys;
keys_list[num_keys + i] = keys + sizeof(SLastKey); keys_list[num_keys + i] = keys + sizeof(SLastKey);
@ -1766,14 +1762,14 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], ((SLastKey *)keys_list[i])->lflag); SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i], values_list_sizes[i]);
taosThreadMutexLock(&pTsdb->rCache.rMutex); taosThreadMutexLock(&pTsdb->rCache.rMutex);
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[i], klen); rocksdb_writebatch_delete(wb, keys_list[i], klen);
} }
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
pLastCol = tsdbCacheDeserialize(values_list[i + num_keys], ((SLastKey *)keys_list[i + num_keys])->lflag); pLastCol = tsdbCacheDeserialize(values_list[i + num_keys], values_list_sizes[i + num_keys]);
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen); rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen);
} }

3
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -385,9 +385,6 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
} }
int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3; int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
if (pr->rowKey.numOfPKs > 0) {
ltype |= CACHESCAN_RETRIEVE_PK;
}
STableKeyInfo* pTableList = pr->pTableList; STableKeyInfo* pTableList = pr->pTableList;

3
source/libs/executor/src/cachescanoperator.c
View File

@ -160,9 +160,6 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
// partition by tbname // partition by tbname
if (oneTableForEachGroup(pTableListInfo) || (totalTables == 1)) { if (oneTableForEachGroup(pTableListInfo) || (totalTables == 1)) {
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL | SCAN_ROW_TYPE(pScanNode->ignoreNull); pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL | SCAN_ROW_TYPE(pScanNode->ignoreNull);
if (pInfo->numOfPks > 0) {
pInfo->retrieveType |= CACHESCAN_RETRIEVE_PK;
}
STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0); STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);