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

opt partition by tbname

This commit is contained in:
liuyao 2023-05-11 18:47:27 +08:00
parent 8277d608f4
commit 8b7b804606
5 changed files with 128 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/stream/tstream.h
View File

@ -239,6 +239,7 @@ typedef struct {
void* vnode; // not available to encoder and decoder
FTbSink* tbSinkFunc;
STSchema* pTSchema;
SSHashObj* pTblInfo;
} STaskSinkTb;
typedef void FSmaSink(void* vnode, int64_t smaId, const SArray* data);

6
source/dnode/vnode/src/tq/tq.c
View File

@ -572,6 +572,10 @@ end:
return ret;
}
void freePtr(void *ptr) {
taosMemoryFree(*(void**)ptr);
}
int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
int32_t vgId = TD_VID(pTq->pVnode);
pTask->id.idStr = createStreamTaskIdStr(pTask->id.streamId, pTask->id.taskId);
@ -646,6 +650,8 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
if (pTask->tbSink.pTSchema == NULL) {
return -1;
}
pTask->tbSink.pTblInfo = tSimpleHashInit(10240, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
tSimpleHashSetFreeFp(pTask->tbSink.pTblInfo, freePtr);
}
if (pTask->taskLevel == TASK_LEVEL__SOURCE) {

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

@ -17,6 +17,13 @@
#include "tmsg.h"
#include "tq.h"
#define MAX_CATCH_NUM 10240
typedef struct STblInfo {
uint64_t uid;
char tbName[TSDB_TABLE_NAME_LEN];
} STblInfo;
int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
const char* pIdStr) {
int32_t totalRows = pDataBlock->info.rows;
@ -90,6 +97,22 @@ end:
return ret;
}
int32_t tqGetTableInfo(SSHashObj* tblInfo ,uint64_t groupId, STblInfo** pTbl) {
void* pVal = tSimpleHashGet(tblInfo, &groupId, sizeof(uint64_t));
if (pVal) {
*pTbl = *(STblInfo**)pVal;
return TSDB_CODE_SUCCESS;
}
return TSDB_CODE_FAILED;
}
int32_t tqPutTableInfo(SSHashObj* tblInfo ,uint64_t groupId, STblInfo* pTbl) {
if (tSimpleHashGetSize(tblInfo) > MAX_CATCH_NUM) {
return TSDB_CODE_SUCCESS;
}
return tSimpleHashPut(tblInfo, &groupId, sizeof(uint64_t), &pTbl, POINTER_BYTES);
}
int32_t tqPutReqToQueue(SVnode* pVnode, SVCreateTbBatchReq* pReqs) {
void* buf = NULL;
int32_t tlen = 0;
@ -260,25 +283,40 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
tbData.suid = suid;
tbData.uid = 0; // uid is assigned by vnode
tbData.sver = pTSchema->version;
STblInfo* pTblMeta = NULL;
char* ctbName = NULL;
tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), pDataBlock->info.parTbName);
if (pDataBlock->info.parTbName[0]) {
ctbName = taosStrdup(pDataBlock->info.parTbName);
} else {
ctbName = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
int32_t res = tqGetTableInfo(pTask->tbSink.pTblInfo, pDataBlock->info.id.groupId, &pTblMeta);
if (res != TSDB_CODE_SUCCESS) {
pTblMeta = taosMemoryCalloc(1, sizeof(STblInfo));
}
char* ctbName = pDataBlock->info.parTbName;
if (!ctbName[0]) {
if (res == TSDB_CODE_SUCCESS) {
memcpy(ctbName, pTblMeta->tbName, strlen(pTblMeta->tbName));
} else {
char* tmp = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
memcpy(ctbName, tmp, strlen(tmp));
memcpy(pTblMeta->tbName, tmp, strlen(tmp));
taosMemoryFree(tmp);
tqDebug("vgId:%d, gropuid:%" PRIu64 " datablock tabel name is null", TD_VID(pVnode),
pDataBlock->info.id.groupId);
}
}
if (res == TSDB_CODE_SUCCESS) {
tbData.uid = pTblMeta->uid;
} else {
SMetaReader mr = {0};
metaReaderInit(&mr, pVnode->pMeta, 0);
if (metaGetTableEntryByName(&mr, ctbName) < 0) {
metaReaderClear(&mr);
taosMemoryFree(pTblMeta);
tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), ctbName);
SVCreateTbReq* pCreateTbReq = NULL;
if (!(pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateStbReq)))) {
taosMemoryFree(ctbName);
goto _end;
};
@ -295,7 +333,6 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
// set tag content
tagArray = taosArrayInit(1, sizeof(STagVal));
if (!tagArray) {
taosMemoryFree(ctbName);
tdDestroySVCreateTbReq(pCreateTbReq);
goto _end;
}
@ -311,11 +348,8 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
tTagNew(tagArray, 1, false, &pTag);
tagArray = taosArrayDestroy(tagArray);
if (pTag == NULL) {
taosMemoryFree(ctbName);
tdDestroySVCreateTbReq(pCreateTbReq);
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(ctbName);
tdDestroySVCreateTbReq(pCreateTbReq);
goto _end;
}
pCreateTbReq->ctb.pTag = (uint8_t*)pTag;
@ -328,8 +362,7 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
pCreateTbReq->ctb.tagName = tagName;
// set table name
pCreateTbReq->name = ctbName;
ctbName = NULL;
pCreateTbReq->name = taosStrdup(ctbName);
tbData.pCreateTbReq = pCreateTbReq;
tbData.flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
@ -338,7 +371,7 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
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);
taosMemoryFree(pTblMeta);
continue;
}
@ -347,13 +380,15 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
", actual suid %" PRId64 "",
TD_VID(pVnode), ctbName, suid, mr.me.ctbEntry.suid);
metaReaderClear(&mr);
taosMemoryFree(ctbName);
taosMemoryFree(pTblMeta);
continue;
}
tbData.uid = mr.me.uid;
pTblMeta->uid = mr.me.uid;
tqPutTableInfo(pTask->tbSink.pTblInfo, pDataBlock->info.id.groupId, pTblMeta);
metaReaderClear(&mr);
taosMemoryFreeClear(ctbName);
}
}
// rows

2
source/libs/executor/src/scanoperator.c
View File

@ -1583,7 +1583,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
// currently only the tbname pseudo column
if (pInfo->numOfPseudoExpr > 0) {
int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,
pInfo->pRes->info.rows, GET_TASKID(pTaskInfo), NULL);
pInfo->pRes->info.rows, GET_TASKID(pTaskInfo), &pTableScanInfo->base.metaCache);
// ignore the table not exists error, since this table may have been dropped during the scan procedure.
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {
blockDataFreeRes((SSDataBlock*)pBlock);

1
source/libs/stream/src/streamTask.c
View File

@ -195,6 +195,7 @@ void tFreeStreamTask(SStreamTask* pTask) {
if (pTask->outputType == TASK_OUTPUT__TABLE) {
tDeleteSchemaWrapper(pTask->tbSink.pSchemaWrapper);
taosMemoryFree(pTask->tbSink.pTSchema);
tSimpleHashCleanup(pTask->tbSink.pTblInfo);
}
if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {