From fbc50a3da38563a032abf3b9ffa83935c698c453 Mon Sep 17 00:00:00 2001 From: Tao Liu Date: Tue, 26 May 2020 11:02:05 +0000 Subject: [PATCH] [TD-90] tag schema develop --- src/common/inc/tdataformat.h | 17 +++-- src/common/src/tdataformat.c | 143 +++++++++++++++++++++++++++++------ src/tsdb/src/tsdbMeta.c | 28 ++++--- src/tsdb/src/tsdbRead.c | 16 ++-- src/vnode/src/vnodeWrite.c | 6 +- 5 files changed, 158 insertions(+), 52 deletions(-) diff --git a/src/common/inc/tdataformat.h b/src/common/inc/tdataformat.h index 0aec948015..b1dda88d50 100644 --- a/src/common/inc/tdataformat.h +++ b/src/common/inc/tdataformat.h @@ -226,13 +226,13 @@ pData typedef struct { int16_t colId; // column ID int16_t colType; - int16_t colLen; // if col type is binary/Nchar, this is the length of binary/Nchar - int16_t offset; //to store value for numeric col or offset for binary/Nchar + uint16_t offset; //to store value for numeric col or offset for binary/Nchar } STagCol; typedef struct { - int32_t len; + int32_t len; void * pData; // Space to store the tag value + uint16_t dataLen; int16_t ncols; // Total columns allocated STagCol tagCols[]; } STagRow; @@ -242,10 +242,13 @@ typedef struct { int tdInsertTagCol(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 -int tdQuerTagByID(SDataRow row, int16_t colId, void *value, int16_t *type, int16_t *len); //if find tag, 0, else return -1; -int tdAppendTagColVal(SDataRow row, void *value, int8_t type, int32_t bytes); - -SDataRow tdNewTagRowFromSchema(STSchema *pSchema); +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 diff --git a/src/common/src/tdataformat.c b/src/common/src/tdataformat.c index 3df94452d8..822a92322a 100644 --- a/src/common/src/tdataformat.c +++ b/src/common/src/tdataformat.c @@ -14,6 +14,7 @@ */ #include "tdataformat.h" #include "wchar.h" +#include "talgo.h" /** * Create a SSchema object with nCols columns @@ -152,25 +153,136 @@ SDataRow tdNewDataRowFromSchema(STSchema *pSchema) { } int tdInsertTagCol(SDataRow row, void *value, int16_t len, int8_t type, int16_t colId){ //insert tag value and update all the information + //todo return 0; }; int tdDeleteTagCol(SDataRow row, int16_t colId){ // delete tag value and update all the information - return o; + //todo + return 0; }; -int tdQuerTagByID(SDataRow row, int16_t colId, void *value, int16_t *type, int16_t *len){ //if find tag, 0, else return -1; - return 0; +static int compTagVal(const void *key1, const void *key2) { + if (*(int16_t *)key1 > *(int16_t *)key2) { + return 1; + } else if (*(int16_t *)key1 == *(int16_t *)key2) { + return 0; + } else { + return -1; + } +} + +void * tdQueryTagByID(SDataRow row, int16_t colId, int16_t *type) { //if find tag, 0, else return -1; + //todo + ASSERT(((STagRow *)row)->pData != NULL); + + STagCol *pBase = ((STagRow *)row)->tagCols; + int16_t nCols = ((STagRow *)row)->ncols; + + STagCol * stCol = taosbsearch(&colId, pBase, nCols, sizeof(STagCol), compTagVal, 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){ +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; }; -SDataRow tdNewTagRowFromSchema(STSchema *pSchema) { - //todo +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 - pSchema->flen; + 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) + sizeof(int32_t); + 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 */ @@ -203,25 +315,6 @@ int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_ return 0; } -int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_t offset) { - ASSERT(value != NULL); - int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE; - char * ptr = POINTER_SHIFT(row, dataRowLen(row)); - - switch (type) { - case TSDB_DATA_TYPE_BINARY: - case TSDB_DATA_TYPE_NCHAR: - *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row); - memcpy(ptr, value, varDataTLen(value)); - dataRowLen(row) += varDataTLen(value); - break; - default: - memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]); - break; - } - - return 0; -} SDataRow tdDataRowDup(SDataRow row) { SDataRow trow = malloc(dataRowLen(row)); diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c index 356e9c77f1..c16bb74df8 100644 --- a/src/tsdb/src/tsdbMeta.c +++ b/src/tsdb/src/tsdbMeta.c @@ -54,7 +54,7 @@ void *tsdbEncodeTable(STable *pTable, int *contLen) { ptr = tdEncodeSchema(ptr, pTable->schema); ptr = tdEncodeSchema(ptr, pTable->tagSchema); } else if (pTable->type == TSDB_CHILD_TABLE) { - dataRowCpy(ptr, pTable->tagVal); + tdTagRowCpy(ptr, pTable->tagVal); } else { ptr = tdEncodeSchema(ptr, pTable->schema); } @@ -96,7 +96,7 @@ STable *tsdbDecodeTable(void *cont, int contLen) { pTable->schema = tdDecodeSchema(&ptr); pTable->tagSchema = tdDecodeSchema(&ptr); } else if (pTable->type == TSDB_CHILD_TABLE) { - pTable->tagVal = tdDataRowDup(ptr); + pTable->tagVal = tdTagRowDecode(ptr); } else { pTable->schema = tdDecodeSchema(&ptr); } @@ -114,8 +114,10 @@ static char* getTagIndexKey(const void* pData) { SDataRow row = elem->pTable->tagVal; STSchema* pSchema = tsdbGetTableTagSchema(elem->pMeta, elem->pTable); STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN]; - - return tdGetRowDataOfCol(row, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset); + int16_t type = 0; + void * res = tdQueryTagByID(row, pCol->colId,&type); + ASSERT(type == pCol->type); + return res; } int tsdbRestoreTable(void *pHandle, void *cont, int contLen) { @@ -255,8 +257,9 @@ int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t } SDataRow row = (SDataRow)pTable->tagVal; - char* d = tdGetRowDataOfCol(row, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset); - + int16_t tagtype = 0; + char* d = tdQueryTagByID(row, pCol->colId, &tagtype); + //ASSERT((int8_t)tagtype == pCol->type) *val = d; *type = pCol->type; *bytes = pCol->bytes; @@ -321,7 +324,7 @@ int tsdbCreateTable(TsdbRepoT *repo, STableCfg *pCfg) { if (super->pIndex == NULL) { tdFreeSchema(super->schema); tdFreeSchema(super->tagSchema); - tdFreeDataRow(super->tagVal); + tdFreeTagRow(super->tagVal); free(super); return -1; } @@ -346,7 +349,7 @@ int tsdbCreateTable(TsdbRepoT *repo, STableCfg *pCfg) { if (IS_CREATE_STABLE(pCfg)) { // TSDB_CHILD_TABLE table->type = TSDB_CHILD_TABLE; table->superUid = pCfg->superUid; - table->tagVal = tdDataRowDup(pCfg->tagValues); + table->tagVal = tdTagRowDup(pCfg->tagValues); } else { // TSDB_NORMAL_TABLE table->type = TSDB_NORMAL_TABLE; table->superUid = -1; @@ -433,7 +436,7 @@ static void tsdbFreeMemTable(SMemTable *pMemTable) { static int tsdbFreeTable(STable *pTable) { // TODO: finish this function if (pTable->type == TSDB_CHILD_TABLE) { - tdFreeDataRow(pTable->tagVal); + tdFreeTagRow(pTable->tagVal); } else { tdFreeSchema(pTable->schema); } @@ -571,7 +574,9 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) { STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable); STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN]; - char* key = tdGetRowDataOfCol(pTable->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset); + int16_t tagtype = 0; + char* key = tdQueryTagByID(pTable->tagVal, pCol->colId, &tagtype); + ASSERT(pCol->type == tagtype); SArray* res = tSkipListGet(pSTable->pIndex, key); size_t size = taosArrayGetSize(res); @@ -602,7 +607,8 @@ static int tsdbEstimateTableEncodeSize(STable *pTable) { size += tdGetSchemaEncodeSize(pTable->schema); size += tdGetSchemaEncodeSize(pTable->tagSchema); } else if (pTable->type == TSDB_CHILD_TABLE) { - size += dataRowLen(pTable->tagVal); + STagRow *pTagRow = (STagRow *)(pTable->tagVal); + size += dataRowLen(pTable->tagVal) + pTagRow->dataLen; } else { size += tdGetSchemaEncodeSize(pTable->schema); } diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index ad3da226f6..4912f6f2e8 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -1746,9 +1746,9 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) { STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex); bytes = pCol->bytes; type = pCol->type; - - f1 = tdGetRowDataOfCol(pTable1->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset); - f2 = tdGetRowDataOfCol(pTable2->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset); + int16_t tgtype1, tgtype2 = 0; + f1 = tdQueryTagByID(pTable1->tagVal, pCol->colId, &tgtype1); + f2 = tdQueryTagByID(pTable2->tagVal, pCol->colId, &tgtype2); } int32_t ret = doCompare(f1, f2, type, bytes); @@ -1836,10 +1836,12 @@ bool indexedNodeFilterFp(const void* pNode, void* param) { val = (char*) elem->pTable->name; type = TSDB_DATA_TYPE_BINARY; } else { - STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema?? - - int32_t offset = pTSchema->columns[pInfo->colIndex].offset; - val = tdGetRowDataOfCol(elem->pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset); +// STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema?? + 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->colIndex, &type); + // ASSERT(pInfo->sch.type == type); } int32_t ret = 0; diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index 9c415d6af7..96ca19e129 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -139,11 +139,13 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe char *pTagData = pTable->data + totalCols * sizeof(SSchema); int accumBytes = 0; - dataRow = tdNewDataRowFromSchema(pDestTagSchema); + //dataRow = tdNewDataRowFromSchema(pDestTagSchema); + dataRow = tdNewTagRowFromSchema(pDestTagSchema, numOfTags); for (int i = 0; i < numOfTags; i++) { STColumn *pTCol = schemaColAt(pDestTagSchema, i); - tdAppendColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->offset); +// tdAppendColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->offset); + tdAppendTagColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->colId); accumBytes += htons(pSchema[i + numOfColumns].bytes); } tsdbTableSetTagValue(&tCfg, dataRow, false); -- GitLab