提交 fbc50a3d 编写于 作者: T Tao Liu

[TD-90] tag schema develop

上级 eb6b6e0c
......@@ -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
......
......@@ -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));
......
......@@ -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);
}
......
......@@ -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;
......
......@@ -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);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册