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

refactor:modify schemaless function to speed

This commit is contained in:
wangmm0220 2022-05-07 16:24:30 +08:00
parent 1025c3c708
commit d418abd07f
3 changed files with 100 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/parser/parser.h
View File

@ -105,7 +105,7 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
void* smlInitHandle(SQuery *pQuery);
void smlDestroyHandle(void *pHandle);
int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *colsHash, SArray *cols, bool format, STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen);
int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SArray *colsSchema, SArray *cols, bool format, STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen);
int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
#ifdef __cplusplus

157
source/client/src/clientSml.c
View File

@ -37,8 +37,8 @@ typedef enum {
typedef struct {
char sTableName[TSDB_TABLE_NAME_LEN];
SHashObj *tags;
SHashObj *fields;
SArray *tags;
SArray *fields;
} SCreateSTableActionInfo;
typedef struct {
@ -78,14 +78,17 @@ typedef struct {
// colsFormat store cols formated, for quick parse, if info->formatData is true
SArray *colsFormat; // elements are SArray<SSmlKv*>
// cols & colsColumn store cols un formated
// cols store cols un formated
SArray *cols; // elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
SHashObj *columnsHash; // elements are <cols key string, 1>, just for judge if key exists quickly.
} SSmlTableInfo;
typedef struct {
SHashObj *tagHash;
SArray *tags; // save the origin order to create table
SHashObj *tagHash; // elements are <key, index in tags>
SArray *cols;
SHashObj *fieldHash;
STableMeta *tableMeta;
} SSmlSTableMeta;
@ -350,37 +353,26 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
int n = sprintf(result, "create stable %s (", action->createSTable.sTableName);
char* pos = result + n; int freeBytes = capacity - n;
size_t size = taosHashGetSize(action->createSTable.fields);
SArray *cols = taosArrayInit(size, POINTER_BYTES);
SSmlKv **kv = taosHashIterate(action->createSTable.fields, NULL);
while(kv){
if(strncmp((*kv)->key, TS, strlen(TS)) == 0 && (*kv)->type == TSDB_DATA_TYPE_TIMESTAMP){
taosArrayInsert(cols, 0, kv);
}else{
taosArrayPush(cols, kv);
}
kv = taosHashIterate(action->createSTable.fields, kv);
}
SArray *cols = action->createSTable.tags;
for(int i = 0; i < taosArrayGetSize(cols); i++){
SSmlKv *kvNew = taosArrayGetP(cols, i);
smlBuildColumnDescription(kvNew, pos, freeBytes, &outBytes);
SSmlKv *kv = taosArrayGetP(cols, i);
smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
pos += outBytes; freeBytes -= outBytes;
*pos = ','; ++pos; --freeBytes;
}
taosArrayDestroy(cols);
--pos; ++freeBytes;
outBytes = snprintf(pos, freeBytes, ") tags (");
pos += outBytes; freeBytes -= outBytes;
kv = taosHashIterate(action->createSTable.tags, NULL);
while(kv){
smlBuildColumnDescription(*kv, pos, freeBytes, &outBytes);
cols = action->createSTable.tags;
for(int i = 0; i < taosArrayGetSize(cols); i++){
SSmlKv *kv = taosArrayGetP(cols, i);
smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
pos += outBytes; freeBytes -= outBytes;
*pos = ','; ++pos; --freeBytes;
kv = taosHashIterate(action->createSTable.tags, kv);
}
pos--; ++freeBytes;
outBytes = snprintf(pos, freeBytes, ")");
@ -419,7 +411,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
SSmlSTableMeta** tableMetaSml = taosHashIterate(info->superTables, NULL);
while (tableMetaSml) {
SSmlSTableMeta* cTablePoints = *tableMetaSml;
SSmlSTableMeta* sTableData = *tableMetaSml;
STableMeta *pTableMeta = NULL;
SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
@ -436,8 +428,8 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
SSchemaAction schemaAction = {0};
schemaAction.action = SCHEMA_ACTION_CREATE_STABLE;
memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
schemaAction.createSTable.tags = cTablePoints->tagHash;
schemaAction.createSTable.fields = cTablePoints->fieldHash;
schemaAction.createSTable.tags = sTableData->tags;
schemaAction.createSTable.fields = sTableData->cols;
code = smlApplySchemaAction(info, &schemaAction);
if (code != 0) {
uError("SML:0x%"PRIx64" smlApplySchemaAction failed. can not create %s", info->id, schemaAction.createSTable.sTableName);
@ -454,7 +446,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
return code;
}
cTablePoints->tableMeta = pTableMeta;
sTableData->tableMeta = pTableMeta;
tableMetaSml = taosHashIterate(info->superTables, tableMetaSml);
}
@ -1295,14 +1287,19 @@ static bool smlUpdateMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols,
SSmlKv *kv = taosArrayGetP(tags, i);
ASSERT(kv->type == TSDB_DATA_TYPE_NCHAR);
SSmlKv **value = taosHashGet(tableMeta->tagHash, kv->key, kv->keyLen);
if(value){
uint8_t *index = taosHashGet(tableMeta->tagHash, kv->key, kv->keyLen);
if(index){
SSmlKv **value = taosArrayGet(tableMeta->tags, *index);
ASSERT((*value)->type == TSDB_DATA_TYPE_NCHAR);
if(kv->valueLen > (*value)->valueLen){ // tags type is nchar
*value = kv;
}
}else{
taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
size_t tmp = taosArrayGetSize(tableMeta->tags);
ASSERT(tmp <= UINT8_MAX);
uint8_t size = tmp;
taosArrayPush(tableMeta->tags, &kv);
taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &size, CHAR_BYTES);
}
}
}
@ -1310,8 +1307,10 @@ static bool smlUpdateMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols,
if(cols){
for (int i = 1; i < taosArrayGetSize(cols); ++i) { //jump timestamp
SSmlKv *kv = taosArrayGetP(cols, i);
SSmlKv **value = taosHashGet(tableMeta->fieldHash, kv->key, kv->keyLen);
if(value){
int16_t *index = taosHashGet(tableMeta->fieldHash, kv->key, kv->keyLen);
if(index){
SSmlKv **value = taosArrayGet(tableMeta->cols, *index);
if(kv->type != (*value)->type){
smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
return false;
@ -1323,7 +1322,11 @@ static bool smlUpdateMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols,
}
}
}else{
taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
size_t tmp = taosArrayGetSize(tableMeta->cols);
ASSERT(tmp <= INT16_MAX);
int16_t size = tmp;
taosArrayPush(tableMeta->cols, &kv);
taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &size, SHORT_BYTES);
}
}
}
@ -1332,16 +1335,18 @@ static bool smlUpdateMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols,
static void smlInsertMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols){
if(tags){
for (int i = 0; i < taosArrayGetSize(tags); ++i) {
for (uint8_t i = 0; i < taosArrayGetSize(tags); ++i) {
SSmlKv *kv = taosArrayGetP(tags, i);
taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
taosArrayPush(tableMeta->tags, &kv);
taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &i, CHAR_BYTES);
}
}
if(cols){
for (int i = 0; i < taosArrayGetSize(cols); ++i) {
for (int16_t i = 0; i < taosArrayGetSize(cols); ++i) {
SSmlKv *kv = taosArrayGetP(cols, i);
taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
taosArrayPush(tableMeta->cols, &kv);
taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &i, SHORT_BYTES);
}
}
}
@ -1364,12 +1369,6 @@ static SSmlTableInfo* smlBuildTableInfo(bool format){
uError("SML:smlParseLine failed to allocate memory");
goto cleanup;
}
tag->columnsHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (tag->columnsHash == NULL) {
uError("SML:smlParseLine failed to allocate memory");
goto cleanup;
}
}
tag->tags = taosArrayInit(16, POINTER_BYTES);
@ -1399,7 +1398,6 @@ static void smlDestroyBuildTableInfo(SSmlTableInfo *tag, bool format){
}
taosHashCleanup(kvHash);
}
taosHashCleanup(tag->columnsHash);
}
taosArrayDestroy(tag->tags);
taosMemoryFreeClear(tag);
@ -1408,23 +1406,20 @@ static void smlDestroyBuildTableInfo(SSmlTableInfo *tag, bool format){
static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
if(dataFormat){
taosArrayPush(oneTable->colsFormat, &cols);
}else{
SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if(!kvHash){
uError("SML:smlDealCols failed to allocate memory");
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
for(size_t i = 0; i < taosArrayGetSize(cols); i++){
SSmlKv *kv = taosArrayGetP(cols, i);
taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES); // todo key need escape, like \=, because find by schema name later
if(taosHashGet(oneTable->columnsHash, kv->key, kv->keyLen) != NULL){
continue;
}
taosHashPut(oneTable->columnsHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
}
taosArrayPush(oneTable->cols, &kvHash);
return TSDB_CODE_SUCCESS;
}
SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if(!kvHash){
uError("SML:smlDealCols failed to allocate memory");
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
for(size_t i = 0; i < taosArrayGetSize(cols); i++){
SSmlKv *kv = taosArrayGetP(cols, i);
taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES); // todo key need escape, like \=, because find by schema name later
}
taosArrayPush(oneTable->cols, &kvHash);
return TSDB_CODE_SUCCESS;
}
@ -1444,6 +1439,18 @@ static SSmlSTableMeta* smlBuildSTableMeta(){
uError("SML:smlBuildSTableMeta failed to allocate memory");
goto cleanup;
}
meta->tags = taosArrayInit(32, POINTER_BYTES);
if (meta->tags == NULL) {
uError("SML:smlBuildSTableMeta failed to allocate memory");
goto cleanup;
}
meta->cols = taosArrayInit(32, POINTER_BYTES);
if (meta->cols == NULL) {
uError("SML:smlBuildSTableMeta failed to allocate memory");
goto cleanup;
}
return meta;
cleanup:
@ -1454,6 +1461,8 @@ cleanup:
static void smlDestroySTableMeta(SSmlSTableMeta *meta){
taosHashCleanup(meta->tagHash);
taosHashCleanup(meta->fieldHash);
taosArrayDestroy(meta->tags);
taosArrayDestroy(meta->cols);
taosMemoryFree(meta->tableMeta);
}
@ -1500,45 +1509,45 @@ static int32_t smlParseLine(SSmlHandle* info, const char* sql) {
return ret;
}
}else{
SSmlTableInfo *tag = smlBuildTableInfo(info->dataFormat);
if(!tag){
SSmlTableInfo *tinfo = smlBuildTableInfo(info->dataFormat);
if(!tinfo){
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
ret = smlDealCols(tag, info->dataFormat, cols);
ret = smlDealCols(tinfo, info->dataFormat, cols);
if(ret != TSDB_CODE_SUCCESS){
return ret;
}
ret = smlParseCols(elements.tags, elements.tagsLen, tag->tags, true, &info->msgBuf);
ret = smlParseCols(elements.tags, elements.tagsLen, tinfo->tags, true, &info->msgBuf);
if(ret != TSDB_CODE_SUCCESS){
uError("SML:0x%"PRIx64" smlParseCols parse tag fields failed", info->id);
return ret;
}
if(taosArrayGetSize(tag->tags) > TSDB_MAX_TAGS){
if(taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS){
smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
return TSDB_CODE_SML_INVALID_DATA;
}
tag->sTableName = elements.measure;
tag->sTableNameLen = elements.measureLen;
smlBuildChildTableName(tag);
uDebug("SML:0x%"PRIx64" child table name: %s", info->id, tag->childTableName);
tinfo->sTableName = elements.measure;
tinfo->sTableNameLen = elements.measureLen;
smlBuildChildTableName(tinfo);
uDebug("SML:0x%"PRIx64" child table name: %s", info->id, tinfo->childTableName);
SSmlSTableMeta** tableMeta = taosHashGet(info->superTables, elements.measure, elements.measureLen);
if(tableMeta){ // update meta
ret = smlUpdateMeta(*tableMeta, tag->tags, cols, &info->msgBuf);
ret = smlUpdateMeta(*tableMeta, tinfo->tags, cols, &info->msgBuf);
if(!ret){
uError("SML:0x%"PRIx64" smlUpdateMeta failed", info->id);
return TSDB_CODE_SML_INVALID_DATA;
}
}else{
SSmlSTableMeta *meta = smlBuildSTableMeta();
smlInsertMeta(meta, tag->tags, cols);
smlInsertMeta(meta, tinfo->tags, cols);
taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
}
taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tag, POINTER_BYTES);
taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tinfo, POINTER_BYTES);
}
return TSDB_CODE_SUCCESS;
}
@ -1651,7 +1660,7 @@ static int32_t smlInsertData(SSmlHandle* info) {
(*pMeta)->tableMeta->vgId = vg.vgId;
(*pMeta)->tableMeta->uid = tableData->uid; // one table merge data block together according uid
code = smlBindData(info->exec, tableData->tags, tableData->colsFormat, tableData->columnsHash,
code = smlBindData(info->exec, tableData->tags, tableData->colsFormat, (*pMeta)->cols,
tableData->cols, info->dataFormat, (*pMeta)->tableMeta, tableData->childTableName, info->msgBuf.buf, info->msgBuf.len);
if(code != TSDB_CODE_SUCCESS){
return code;
@ -1730,7 +1739,7 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
return NULL;
}
SSmlHandle* info = smlBuildSmlInfo(taos, request, protocol, precision, false);
SSmlHandle* info = smlBuildSmlInfo(taos, request, protocol, precision, true);
if(!info){
return (TAOS_RES*)request;
}

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

@ -1642,7 +1642,7 @@ static int32_t smlBoundTags(SArray *cols, SKVRowBuilder *tagsBuilder, SParsedDat
return TSDB_CODE_SUCCESS;
}
int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *colsHash, SArray *cols, bool format,
int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SArray *colsSchema, SArray *cols, bool format,
STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen) {
SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
@ -1673,21 +1673,7 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *co
SSchema* pSchema = getTableColumnSchema(pTableMeta);
if(format){
ret = smlBoundColumns(taosArrayGetP(colsFormat, 0), &pDataBlock->boundColumnInfo, pSchema);
}else{
SArray *columns = taosArrayInit(16, POINTER_BYTES);
void **p1 = taosHashIterate(colsHash, NULL);
while (p1) {
SSmlKv* kv = *p1;
taosArrayPush(columns, &kv);
p1 = taosHashIterate(colsHash, p1);
}
ret = smlBoundColumns(columns, &pDataBlock->boundColumnInfo, pSchema);
taosArrayDestroy(columns);
}
ret = smlBoundColumns(colsSchema, &pDataBlock->boundColumnInfo, pSchema);
if(ret != TSDB_CODE_SUCCESS){
buildInvalidOperationMsg(&pBuf, "bound cols error");
return ret;
@ -1712,8 +1698,10 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *co
STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size); // skip the SSubmitBlk header
tdSRowResetBuf(pBuilder, row);
void *rowData = NULL;
bool eleEqual = false;
if(format){
rowData = taosArrayGetP(colsFormat, r);
eleEqual = (taosArrayGetSize(rowData) == spd->numOfBound);
}else{
rowData = taosArrayGetP(cols, r);
}
@ -1728,17 +1716,22 @@ int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *co
SSmlKv *kv = NULL;
if(format){
kv = taosArrayGetP(rowData, c);
if (!kv){
char msg[64] = {0};
sprintf(msg, "cols num not the same like before:%d", r);
return buildInvalidOperationMsg(&pBuf, msg);
}
do{
if (!eleEqual && kv && (kv->keyLen != strlen(pColSchema->name) || strncmp(kv->key, pColSchema->name, kv->keyLen) != 0)){
MemRowAppend(&pBuf, NULL, 0, &param);
c++;
if(c >= spd->numOfBound) break;
pColSchema = &pSchema[spd->boundColumns[c] - 1];
continue;
}
break;
}while(1);
}else{
void **p =taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
kv = *p;
if(p) kv = *p;
}
if (kv->length == 0) {
if (!kv || kv->length == 0) {
MemRowAppend(&pBuf, NULL, 0, &param);
} else {
int32_t colLen = pColSchema->bytes;