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

refactor tag function

This commit is contained in:
Hongze Cheng 2020-06-01 06:20:24 +00:00
parent a0f3a9e268
commit 4a1f43c983
8 changed files with 21 additions and 232 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/common/inc/tdataformat.h
View File

@ -248,6 +248,7 @@ typedef struct {
#define kvRowCpy(dst, r) memcpy((dst), (r), kvRowLen(r)) #define kvRowCpy(dst, r) memcpy((dst), (r), kvRowLen(r))
#define kvRowColVal(r, colIdx) POINTER_SHIFT(kvRowValues(r), (colIdx)->offset) #define kvRowColVal(r, colIdx) POINTER_SHIFT(kvRowValues(r), (colIdx)->offset)
#define kvRowColIdxAt(r, i) (kvRowColIdx(r) + (i)) #define kvRowColIdxAt(r, i) (kvRowColIdx(r) + (i))
#define kvRowFree(r) tfree(r)
SKVRow tdKVRowDup(SKVRow row); SKVRow tdKVRowDup(SKVRow row);
SKVRow tdSetKVRowDataOfCol(SKVRow row, int16_t colId, int8_t type, void *value); SKVRow tdSetKVRowDataOfCol(SKVRow row, int16_t colId, int8_t type, void *value);
@ -314,57 +315,6 @@ static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId,
return 0; return 0;
} }
// ----------------- Tag row structure
/* A tag row, the format is like below:
+----------+----------------------------------------------------------------+
| STagRow | STagCol | STagCol | STagCol | STagCol | ...| STagCol | STagCol |
+----------+----------------------------------------------------------------+
pData
+----------+----------------------------------------------------------------+
| value 1 | value 2 | value 3 | value 4 | ....|value n |
+----------+----------------------------------------------------------------+
*/
#define TD_TAG_ROW_HEAD_SIZE sizeof(int16_t)
#define tagRowNum(r) (*(int16_t *)(r))
#define tagRowArray(r) POINTER_SHIFT(r, TD_TAG_ROW_HEAD_SIZE)
//#define dataRowKey(r) (*(TSKEY *)(dataRowTuple(r)))
//#define dataRowSetLen(r, l) (dataRowLen(r) = (l))
//#define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
//#define dataRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_DATA_ROW_HEAD_SIZE)
typedef struct {
int16_t colId; // column ID
int16_t colType;
uint16_t offset; // to store value for numeric col or offset for binary/Nchar
} STagCol;
typedef struct {
int32_t len;
void * pData; // Space to store the tag value
uint16_t dataLen;
int16_t ncols; // Total columns allocated
STagCol tagCols[];
} STagRow;
#define tagColSize(r) (sizeof(STagCol) + r.colLen)
int tdSetTagCol(SDataRow row, void *value, int16_t len, int8_t type,
int16_t colId); // insert tag value and update all the information
int tdDeleteTagCol(SDataRow row, int16_t colId); // delete tag value and update all the information
void * tdQueryTagByID(SDataRow row, int16_t colId, int16_t *type); // if find tag, 0, else return -1;
int tdAppendTagColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int16_t colId);
SDataRow tdTagRowDup(SDataRow row);
void tdFreeTagRow(SDataRow row);
SDataRow tdTagRowDecode(SDataRow row);
int tdTagRowCpy(SDataRow dst, SDataRow src);
void * tdNewTagRowFromSchema(STSchema *pSchema, int16_t numofTags);
STSchema *tdGetSchemaFromData(SDataRow *row);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

146
src/common/src/tdataformat.c
View File

@ -152,152 +152,6 @@ SDataRow tdNewDataRowFromSchema(STSchema *pSchema) {
return row; return row;
} }
int tdSetTagCol(SDataRow row, void *value, int16_t len, int8_t type,
int16_t colId) { // insert/update tag value and update all the information
ASSERT(((STagRow *)row)->pData != NULL);
// STagCol * stCol = tdQueryTagColByID()
return 0;
};
int tdDeleteTagCol(SDataRow row, int16_t colId) { // delete tag value and update all the information
// todo
return 0;
};
static int compTagId(const void *key1, const void *key2) {
if (((STagCol *)key1)->colId > ((STagCol *)key2)->colId) {
return 1;
} else if (((STagCol *)key1)->colId == ((STagCol *)key2)->colId) {
return 0;
} else {
return -1;
}
}
/**
* Find tag structure by colId, if find, return tag structure, else return NULL;
*/
STagCol *tdQueryTagColByID(SDataRow row, int16_t colId, int flags) { // if find tag, 0, else return -1;
ASSERT(((STagRow *)row)->pData != NULL);
STagCol *pBase = ((STagRow *)row)->tagCols;
int16_t nCols = ((STagRow *)row)->ncols;
STagCol key = {colId, 0, 0};
STagCol *stCol = taosbsearch(&key, pBase, nCols, sizeof(STagCol), compTagId, flags);
return stCol;
};
/**
* Find tag value by colId, if find, return tag value, else return NULL;
*/
void *tdQueryTagByID(SDataRow row, int16_t colId, int16_t *type) {
ASSERT(((STagRow *)row)->pData != NULL);
STagCol *pBase = ((STagRow *)row)->tagCols;
int16_t nCols = ((STagRow *)row)->ncols;
STagCol key = {colId, 0, 0};
STagCol *stCol = taosbsearch(&key, pBase, nCols, sizeof(STagCol), compTagId, TD_EQ);
if (NULL == stCol) {
return NULL;
}
void *pData = ((STagRow *)row)->pData;
*type = stCol->colType;
return pData + stCol->offset;
};
int tdAppendTagColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int16_t colId) {
ASSERT(value != NULL);
// ASSERT(bytes-2 == varDataTLen(value));
ASSERT(row != NULL);
STagRow *pTagrow = row;
pTagrow->tagCols[pTagrow->ncols].colId = colId;
pTagrow->tagCols[pTagrow->ncols].colType = type;
pTagrow->tagCols[pTagrow->ncols].offset = pTagrow->dataLen;
switch (type) {
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memcpy((char *)pTagrow->pData + pTagrow->dataLen, value, varDataTLen(value));
pTagrow->dataLen += varDataTLen(value);
break;
default:
memcpy((char *)pTagrow->pData + pTagrow->dataLen, value, TYPE_BYTES[type]);
pTagrow->dataLen += TYPE_BYTES[type];
break;
}
pTagrow->ncols++;
return 0;
};
void *tdNewTagRowFromSchema(STSchema *pSchema, int16_t numofTags) {
int32_t size = sizeof(STagRow) + numofTags * sizeof(STagCol);
STagRow *row = malloc(size);
if (row == NULL) return NULL;
int32_t datasize = pSchema->tlen;
row->pData = malloc(datasize);
if (NULL == row->pData) {
free(row);
return NULL;
}
row->len = size;
row->dataLen = 0;
row->ncols = 0;
return row;
}
/**
* free tag row
*/
void tdFreeTagRow(SDataRow row) {
if (row) {
free(((STagRow *)row)->pData);
free(row);
}
}
SDataRow tdTagRowDup(SDataRow row) {
STagRow *trow = malloc(dataRowLen(row));
if (trow == NULL) return NULL;
dataRowCpy(trow, row);
trow->pData = malloc(trow->dataLen);
if (NULL == trow->pData) {
free(trow);
return NULL;
}
memcpy(trow->pData, ((STagRow *)row)->pData, trow->dataLen);
return trow;
}
SDataRow tdTagRowDecode(SDataRow row) {
STagRow *trow = malloc(dataRowLen(row));
if (trow == NULL) return NULL;
dataRowCpy(trow, row);
trow->pData = malloc(trow->dataLen);
if (NULL == trow->pData) {
free(trow);
return NULL;
}
char *pData = (char *)row + dataRowLen(row);
memcpy(trow->pData, pData, trow->dataLen);
return trow;
}
int tdTagRowCpy(SDataRow dst, SDataRow src) {
if (src == NULL) return -1;
dataRowCpy(dst, src);
void *pData = dst + dataRowLen(src);
memcpy(pData, ((STagRow *)src)->pData, ((STagRow *)src)->dataLen);
return 0;
}
/** /**
* Free the SDataRow object * Free the SDataRow object
*/ */

2
src/inc/tsdb.h
View File

@ -101,7 +101,7 @@ int tsdbInitTableCfg(STableCfg *config, ETableType type, uint64_t uid, int32_t
int tsdbTableSetSuperUid(STableCfg *config, uint64_t uid); int tsdbTableSetSuperUid(STableCfg *config, uint64_t uid);
int tsdbTableSetSchema(STableCfg *config, STSchema *pSchema, bool dup); int tsdbTableSetSchema(STableCfg *config, STSchema *pSchema, bool dup);
int tsdbTableSetTagSchema(STableCfg *config, STSchema *pSchema, bool dup); int tsdbTableSetTagSchema(STableCfg *config, STSchema *pSchema, bool dup);
int tsdbTableSetTagValue(STableCfg *config, SDataRow row, bool dup); int tsdbTableSetTagValue(STableCfg *config, SKVRow row, bool dup);
int tsdbTableSetName(STableCfg *config, char *name, bool dup); int tsdbTableSetName(STableCfg *config, char *name, bool dup);
int tsdbTableSetSName(STableCfg *config, char *sname, bool dup); int tsdbTableSetSName(STableCfg *config, char *sname, bool dup);
int tsdbTableSetStreamSql(STableCfg *config, char *sql, bool dup); int tsdbTableSetStreamSql(STableCfg *config, char *sql, bool dup);

2
src/tsdb/inc/tsdbMain.h
View File

@ -76,7 +76,7 @@ typedef struct STable {
int32_t sversion; int32_t sversion;
STSchema * schema; STSchema * schema;
STSchema * tagSchema; STSchema * tagSchema;
SDataRow tagVal; SKVRow tagVal;
SMemTable * mem; SMemTable * mem;
SMemTable * imem; SMemTable * imem;
void * pIndex; // For TSDB_SUPER_TABLE, it is the skiplist index void * pIndex; // For TSDB_SUPER_TABLE, it is the skiplist index

6
src/tsdb/src/tsdbMain.c
View File

@ -506,11 +506,11 @@ int tsdbTableSetTagSchema(STableCfg *config, STSchema *pSchema, bool dup) {
return 0; return 0;
} }
int tsdbTableSetTagValue(STableCfg *config, SDataRow row, bool dup) { int tsdbTableSetTagValue(STableCfg *config, SKVRow row, bool dup) {
if (config->type != TSDB_CHILD_TABLE) return -1; if (config->type != TSDB_CHILD_TABLE) return -1;
if (dup) { if (dup) {
config->tagValues = tdDataRowDup(row); config->tagValues = tdKVRowDup(row);
} else { } else {
config->tagValues = row; config->tagValues = row;
} }
@ -557,7 +557,7 @@ int tsdbTableSetStreamSql(STableCfg *config, char *sql, bool dup) {
void tsdbClearTableCfg(STableCfg *config) { void tsdbClearTableCfg(STableCfg *config) {
if (config->schema) tdFreeSchema(config->schema); if (config->schema) tdFreeSchema(config->schema);
if (config->tagSchema) tdFreeSchema(config->tagSchema); if (config->tagSchema) tdFreeSchema(config->tagSchema);
if (config->tagValues) tdFreeDataRow(config->tagValues); if (config->tagValues) kvRowFree(config->tagValues);
tfree(config->name); tfree(config->name);
tfree(config->sname); tfree(config->sname);
tfree(config->sql); tfree(config->sql);

23
src/tsdb/src/tsdbMeta.c
View File

@ -47,8 +47,7 @@ void tsdbEncodeTable(STable *pTable, char *buf, int *contLen) {
ptr = tdEncodeSchema(ptr, pTable->schema); ptr = tdEncodeSchema(ptr, pTable->schema);
ptr = tdEncodeSchema(ptr, pTable->tagSchema); ptr = tdEncodeSchema(ptr, pTable->tagSchema);
} else if (pTable->type == TSDB_CHILD_TABLE) { } else if (pTable->type == TSDB_CHILD_TABLE) {
tdTagRowCpy(ptr, pTable->tagVal); ptr = tdEncodeKVRow(ptr, pTable->tagVal);
ptr = POINTER_SHIFT(ptr, dataRowLen(pTable->tagVal) + ((STagRow *)pTable->tagVal)->dataLen);
} else { } else {
ptr = tdEncodeSchema(ptr, pTable->schema); ptr = tdEncodeSchema(ptr, pTable->schema);
} }
@ -94,8 +93,7 @@ STable *tsdbDecodeTable(void *cont, int contLen) {
pTable->schema = tdDecodeSchema(&ptr); pTable->schema = tdDecodeSchema(&ptr);
pTable->tagSchema = tdDecodeSchema(&ptr); pTable->tagSchema = tdDecodeSchema(&ptr);
} else if (pTable->type == TSDB_CHILD_TABLE) { } else if (pTable->type == TSDB_CHILD_TABLE) {
pTable->tagVal = tdTagRowDecode(ptr); ptr = tdDecodeKVRow(ptr, &pTable->tagVal);
ptr = POINTER_SHIFT(ptr, dataRowLen(pTable->tagVal) + ((STagRow *)pTable->tagVal)->dataLen);
} else { } else {
pTable->schema = tdDecodeSchema(&ptr); pTable->schema = tdDecodeSchema(&ptr);
} }
@ -114,12 +112,9 @@ void tsdbFreeEncode(void *cont) {
static char* getTagIndexKey(const void* pData) { static char* getTagIndexKey(const void* pData) {
STableIndexElem* elem = (STableIndexElem*) pData; STableIndexElem* elem = (STableIndexElem*) pData;
SDataRow row = elem->pTable->tagVal;
STSchema* pSchema = tsdbGetTableTagSchema(elem->pMeta, elem->pTable); STSchema* pSchema = tsdbGetTableTagSchema(elem->pMeta, elem->pTable);
STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN]; STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN];
int16_t type = 0; void * res = tdGetKVRowValOfCol(elem->pTable->tagVal, pCol->colId);
void * res = tdQueryTagByID(row, pCol->colId,&type);
ASSERT(type == pCol->type);
return res; return res;
} }
@ -271,9 +266,7 @@ int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t
return -1; // No matched tags. Maybe the modification of tags has not been done yet. return -1; // No matched tags. Maybe the modification of tags has not been done yet.
} }
SDataRow row = (SDataRow)pTable->tagVal; char* d = tdGetKVRowValOfCol(pTable->tagVal, pCol->colId);
int16_t tagtype = 0;
char* d = tdQueryTagByID(row, pCol->colId, &tagtype);
//ASSERT((int8_t)tagtype == pCol->type) //ASSERT((int8_t)tagtype == pCol->type)
*val = d; *val = d;
*type = pCol->type; *type = pCol->type;
@ -352,7 +345,7 @@ static STable *tsdbNewTable(STableCfg *pCfg, bool isSuper) {
if (pCfg->type == TSDB_CHILD_TABLE) { if (pCfg->type == TSDB_CHILD_TABLE) {
pTable->superUid = pCfg->superUid; pTable->superUid = pCfg->superUid;
pTable->tagVal = tdDataRowDup(pCfg->tagValues); pTable->tagVal = tdKVRowDup(pCfg->tagValues);
} else if (pCfg->type == TSDB_NORMAL_TABLE) { } else if (pCfg->type == TSDB_NORMAL_TABLE) {
pTable->superUid = -1; pTable->superUid = -1;
pTable->schema = tdDupSchema(pCfg->schema); pTable->schema = tdDupSchema(pCfg->schema);
@ -487,7 +480,7 @@ static int tsdbFreeTable(STable *pTable) {
if (pTable == NULL) return 0; if (pTable == NULL) return 0;
if (pTable->type == TSDB_CHILD_TABLE) { if (pTable->type == TSDB_CHILD_TABLE) {
tdFreeTagRow(pTable->tagVal); kvRowFree(pTable->tagVal);
} else { } else {
tdFreeSchema(pTable->schema); tdFreeSchema(pTable->schema);
} }
@ -636,9 +629,7 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable); STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable);
STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN]; STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN];
int16_t tagtype = 0; char* key = tdGetKVRowValOfCol(pTable->tagVal, pCol->colId);
char* key = tdQueryTagByID(pTable->tagVal, pCol->colId, &tagtype);
ASSERT(pCol->type == tagtype);
SArray* res = tSkipListGet(pSTable->pIndex, key); SArray* res = tSkipListGet(pSTable->pIndex, key);
size_t size = taosArrayGetSize(res); size_t size = taosArrayGetSize(res);

12
src/tsdb/src/tsdbRead.c
View File

@ -1753,9 +1753,8 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex); STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex);
bytes = pCol->bytes; bytes = pCol->bytes;
type = pCol->type; type = pCol->type;
int16_t tgtype1, tgtype2 = 0; f1 = tdGetKVRowValOfCol(pTable1->tagVal, pCol->colId);
f1 = tdQueryTagByID(pTable1->tagVal, pCol->colId, &tgtype1); f2 = tdGetKVRowValOfCol(pTable2->tagVal, pCol->colId);
f2 = tdQueryTagByID(pTable2->tagVal, pCol->colId, &tgtype2);
} }
int32_t ret = doCompare(f1, f2, type, bytes); int32_t ret = doCompare(f1, f2, type, bytes);
@ -1843,12 +1842,7 @@ bool indexedNodeFilterFp(const void* pNode, void* param) {
val = (char*) elem->pTable->name; val = (char*) elem->pTable->name;
type = TSDB_DATA_TYPE_BINARY; type = TSDB_DATA_TYPE_BINARY;
} else { } else {
// STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema?? val = tdGetKVRowValOfCol(elem->pTable->tagVal, pInfo->sch.colId);
int16_t type;
// int32_t offset = pTSchema->columns[pInfo->colIndex].offset;
// val = tdGetRowDataOfCol(elem->pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
val = tdQueryTagByID(elem->pTable->tagVal, pInfo->sch.colId, &type);
// ASSERT(pInfo->sch.type == type);
} }
//todo :the val is possible to be null, so check it out carefully //todo :the val is possible to be null, so check it out carefully
int32_t ret = 0; int32_t ret = 0;

10
src/vnode/src/vnodeWrite.c
View File

@ -139,16 +139,16 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
char *pTagData = pTable->data + totalCols * sizeof(SSchema); char *pTagData = pTable->data + totalCols * sizeof(SSchema);
int accumBytes = 0; int accumBytes = 0;
//dataRow = tdNewDataRowFromSchema(pDestTagSchema);
dataRow = tdNewTagRowFromSchema(pDestTagSchema, numOfTags);
SKVRowBuilder kvRowBuilder;
tdInitKVRowBuilder(&kvRowBuilder);
for (int i = 0; i < numOfTags; i++) { for (int i = 0; i < numOfTags; i++) {
STColumn *pTCol = schemaColAt(pDestTagSchema, i); STColumn *pTCol = schemaColAt(pDestTagSchema, i);
// tdAppendColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->offset); tdAddColToKVRow(&kvRowBuilder, pTCol->colId, pTCol->type, pTagData + accumBytes);
tdAppendTagColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->colId);
accumBytes += htons(pSchema[i + numOfColumns].bytes); accumBytes += htons(pSchema[i + numOfColumns].bytes);
} }
tsdbTableSetTagValue(&tCfg, dataRow, false); tsdbTableSetTagValue(&tCfg, tdGetKVRowFromBuilder(&kvRowBuilder), false);
tdDestroyKVRowBuilder(&kvRowBuilder);
} }
// only normal has sql string // only normal has sql string