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

refactor(stream): optimize auto create child table

This commit is contained in:
Liu Jicong 2022-10-17 17:18:57 +08:00
parent ced97588d1
commit bcae6e1f4e
1 changed files with 105 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

176
source/dnode/vnode/src/tq/tqSink.c
View File

@ -293,7 +293,6 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
SSchemaWrapper* pSchemaWrapper = pTask->tbSink.pSchemaWrapper; SSchemaWrapper* pSchemaWrapper = pTask->tbSink.pSchemaWrapper;
int32_t blockSz = taosArrayGetSize(pBlocks); int32_t blockSz = taosArrayGetSize(pBlocks);
bool createTb = true;
SArray* tagArray = taosArrayInit(1, sizeof(STagVal)); SArray* tagArray = taosArrayInit(1, sizeof(STagVal));
if (!tagArray) { if (!tagArray) {
@ -303,6 +302,7 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
tqDebug("vgId:%d, task %d write into table, block num: %d", TD_VID(pVnode), pTask->taskId, blockSz); tqDebug("vgId:%d, task %d write into table, block num: %d", TD_VID(pVnode), pTask->taskId, blockSz);
for (int32_t i = 0; i < blockSz; i++) { for (int32_t i = 0; i < blockSz; i++) {
bool createTb = true;
SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i); SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
if (pDataBlock->info.type == STREAM_DELETE_RESULT) { if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
SBatchDeleteReq deleteReq = {0}; SBatchDeleteReq deleteReq = {0};
@ -337,83 +337,115 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
tqDebug("failed to put delete req into write-queue since %s", terrstr()); tqDebug("failed to put delete req into write-queue since %s", terrstr());
} }
} else { } else {
SVCreateTbReq createTbReq = {0}; char* ctbName = NULL;
// set child table name
// set const
createTbReq.flags = 0;
createTbReq.type = TSDB_CHILD_TABLE;
createTbReq.ctb.suid = suid;
// set super table name
SName name = {0};
tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
createTbReq.ctb.stbName = strdup((char*)tNameGetTableName(&name)); // strdup(stbFullName);
// set tag content
taosArrayClear(tagArray);
STagVal tagVal = {
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
};
taosArrayPush(tagArray, &tagVal);
createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);
STag* pTag = NULL;
tTagNew(tagArray, 1, false, &pTag);
if (pTag == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosArrayDestroy(tagArray);
tdDestroySVCreateTbReq(&createTbReq);
return;
}
createTbReq.ctb.pTag = (uint8_t*)pTag;
// set tag name
SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
char tagNameStr[TSDB_COL_NAME_LEN] = {0};
strcpy(tagNameStr, "group_id");
taosArrayPush(tagName, tagNameStr);
createTbReq.ctb.tagName = tagName;
// set table name
if (pDataBlock->info.parTbName[0]) { if (pDataBlock->info.parTbName[0]) {
createTbReq.name = strdup(pDataBlock->info.parTbName); ctbName = strdup(pDataBlock->info.parTbName);
} else { } else {
createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId); ctbName = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
} }
int32_t schemaLen; int32_t schemaLen = 0;
int32_t code; void* schemaStr = NULL;
tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
if (code < 0) {
tdDestroySVCreateTbReq(&createTbReq);
taosArrayDestroy(tagArray);
return;
}
// save schema str int64_t uid = 0;
void* schemaStr = taosMemoryMalloc(schemaLen); SMetaReader mr = {0};
if (schemaStr == NULL) { metaReaderInit(&mr, pVnode->pMeta, 0);
terrno = TSDB_CODE_OUT_OF_MEMORY; if (metaGetTableEntryByName(&mr, ctbName) < 0) {
tdDestroySVCreateTbReq(&createTbReq); tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), ctbName);
taosArrayDestroy(tagArray);
return;
}
SEncoder encoder = {0}; SVCreateTbReq createTbReq = {0};
tEncoderInit(&encoder, schemaStr, schemaLen);
code = tEncodeSVCreateTbReq(&encoder, &createTbReq); // set const
if (code < 0) { createTbReq.flags = 0;
terrno = TSDB_CODE_OUT_OF_MEMORY; createTbReq.type = TSDB_CHILD_TABLE;
tdDestroySVCreateTbReq(&createTbReq); createTbReq.ctb.suid = suid;
taosArrayDestroy(tagArray);
// set super table name
SName name = {0};
tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
createTbReq.ctb.stbName = strdup((char*)tNameGetTableName(&name)); // strdup(stbFullName);
createTbReq.name = ctbName;
ctbName = NULL;
// set tag content
taosArrayClear(tagArray);
STagVal tagVal = {
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
};
taosArrayPush(tagArray, &tagVal);
createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);
STag* pTag = NULL;
tTagNew(tagArray, 1, false, &pTag);
if (pTag == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosArrayDestroy(tagArray);
tdDestroySVCreateTbReq(&createTbReq);
return;
}
createTbReq.ctb.pTag = (uint8_t*)pTag;
// set tag name
SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
char tagNameStr[TSDB_COL_NAME_LEN] = {0};
strcpy(tagNameStr, "group_id");
taosArrayPush(tagName, tagNameStr);
createTbReq.ctb.tagName = tagName;
int32_t code;
tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
if (code < 0) {
tdDestroySVCreateTbReq(&createTbReq);
taosArrayDestroy(tagArray);
return;
}
// set schema str
schemaStr = taosMemoryMalloc(schemaLen);
if (schemaStr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tdDestroySVCreateTbReq(&createTbReq);
taosArrayDestroy(tagArray);
return;
}
SEncoder encoder = {0};
tEncoderInit(&encoder, schemaStr, schemaLen);
code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
if (code < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tdDestroySVCreateTbReq(&createTbReq);
taosArrayDestroy(tagArray);
tEncoderClear(&encoder);
taosMemoryFree(schemaStr);
return;
}
tEncoderClear(&encoder); tEncoderClear(&encoder);
taosMemoryFree(schemaStr); tdDestroySVCreateTbReq(&createTbReq);
return; } else {
if (mr.me.type != TSDB_CHILD_TABLE) {
tqError("vgId:%d, failed to write into %s, since table type incorrect, type %d", TD_VID(pVnode), ctbName,
mr.me.type);
metaReaderClear(&mr);
taosMemoryFree(ctbName);
continue;
}
if (mr.me.ctbEntry.suid != suid) {
tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid: %ld, actual suid %ld",
TD_VID(pVnode), ctbName, suid, mr.me.ctbEntry);
metaReaderClear(&mr);
taosMemoryFree(ctbName);
continue;
}
createTb = false;
uid = mr.me.uid;
metaReaderClear(&mr);
taosMemoryFreeClear(ctbName);
} }
tEncoderClear(&encoder);
tdDestroySVCreateTbReq(&createTbReq);
int32_t cap = sizeof(SSubmitReq); int32_t cap = sizeof(SSubmitReq);
@ -445,9 +477,11 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
memcpy(blkSchema, schemaStr, schemaLen); memcpy(blkSchema, schemaStr, schemaLen);
blkHead->schemaLen = htonl(schemaLen); blkHead->schemaLen = htonl(schemaLen);
rowData = POINTER_SHIFT(blkSchema, schemaLen); rowData = POINTER_SHIFT(blkSchema, schemaLen);
} else {
blkHead->uid = uid;
} }
taosMemoryFree(schemaStr); taosMemoryFreeClear(schemaStr);
for (int32_t j = 0; j < rows; j++) { for (int32_t j = 0; j < rows; j++) {
SRowBuilder rb = {0}; SRowBuilder rb = {0};