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

refactor(tmq): rewrite tq read function

This commit is contained in:
Liu Jicong 2022-04-21 14:30:37 +08:00
parent a98bf9d1b5
commit 3dd3ad1e05
3 changed files with 129 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/dnode/vnode/inc/vnode.h
View File

@ -109,8 +109,7 @@ int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList
int tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList); int tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver); int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle); bool tqNextDataBlock(STqReadHandle *pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo); int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, int32_t *pNumOfRows);
SArray *tqRetrieveDataBlock(STqReadHandle *pHandle);
// need to reposition // need to reposition

11
source/dnode/vnode/src/tq/tqRead.c
View File

@ -82,16 +82,7 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
return false; return false;
} }
int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) { int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, int32_t* pNumOfRows) {
// currently only rows are used
pBlockInfo->numOfCols = taosArrayGetSize(pHandle->pColIdList);
pBlockInfo->rows = pHandle->pBlock->numOfRows;
// pBlockInfo->uid = pHandle->pBlock->uid; // the uid can not be assigned to pBlockData.
return 0;
}
int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, int16_t* pGroupId, int32_t* pNumOfRows) {
/*int32_t sversion = pHandle->pBlock->sversion;*/ /*int32_t sversion = pHandle->pBlock->sversion;*/
// TODO set to real sversion // TODO set to real sversion
int32_t sversion = 0; int32_t sversion = 0;

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

@ -30,12 +30,11 @@
#include "query.h" #include "query.h"
#include "tcompare.h" #include "tcompare.h"
#include "thash.h" #include "thash.h"
#include "vnode.h"
#include "ttypes.h" #include "ttypes.h"
#include "vnode.h"
#define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN) #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
#define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC)) #define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) { void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
for (int32_t i = 0; i < numOfOutput; ++i) { for (int32_t i = 0; i < numOfOutput; ++i) {
@ -90,7 +89,7 @@ static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t ord
} }
int64_t key = tw->skey, interval = pInterval->interval; int64_t key = tw->skey, interval = pInterval->interval;
//convert key to second // convert key to second
key = convertTimePrecision(key, pInterval->precision, TSDB_TIME_PRECISION_MILLI) / 1000; key = convertTimePrecision(key, pInterval->precision, TSDB_TIME_PRECISION_MILLI) / 1000;
if (pInterval->intervalUnit == 'y') { if (pInterval->intervalUnit == 'y') {
@ -98,7 +97,7 @@ static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t ord
} }
struct tm tm; struct tm tm;
time_t t = (time_t)key; time_t t = (time_t)key;
taosLocalTime(&t, &tm); taosLocalTime(&t, &tm);
int mon = (int)(tm.tm_year * 12 + tm.tm_mon + interval * factor); int mon = (int)(tm.tm_year * 12 + tm.tm_mon + interval * factor);
@ -125,8 +124,8 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
// todo handle the time range case // todo handle the time range case
TSKEY sk = INT64_MIN; TSKEY sk = INT64_MIN;
TSKEY ek = INT64_MAX; TSKEY ek = INT64_MAX;
// TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey); // TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey);
// TSKEY ek = MAX(pQueryAttr->window.skey, pQueryAttr->window.ekey); // TSKEY ek = MAX(pQueryAttr->window.skey, pQueryAttr->window.ekey);
if (true) { if (true) {
getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w); getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w);
@ -136,7 +135,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
return true; return true;
} }
while(1) { // todo handle the desc order scan case while (1) { // todo handle the desc order scan case
getNextTimeWindow(pInterval, &w, TSDB_ORDER_ASC); getNextTimeWindow(pInterval, &w, TSDB_ORDER_ASC);
if (w.skey > pBlockInfo->window.ekey) { if (w.skey > pBlockInfo->window.ekey) {
break; break;
@ -148,31 +147,31 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
} }
} }
} else { } else {
// getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w); // getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w);
// assert(w.skey <= pBlockInfo->window.ekey); // assert(w.skey <= pBlockInfo->window.ekey);
// //
// if (w.skey > pBlockInfo->window.skey) { // if (w.skey > pBlockInfo->window.skey) {
// return true; // return true;
// } // }
// //
// while(1) { // while(1) {
// getNextTimeWindow(pQueryAttr, &w); // getNextTimeWindow(pQueryAttr, &w);
// if (w.ekey < pBlockInfo->window.skey) { // if (w.ekey < pBlockInfo->window.skey) {
// break; // break;
// } // }
// //
// assert(w.skey < pBlockInfo->window.skey); // assert(w.skey < pBlockInfo->window.skey);
// if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) { // if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
// return true; // return true;
// } // }
// } // }
} }
return false; return false;
} }
int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) { int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableScanInfo* pInfo = pOperator->info; STableScanInfo* pInfo = pOperator->info;
STaskCostInfo* pCost = &pTaskInfo->cost; STaskCostInfo* pCost = &pTaskInfo->cost;
@ -189,13 +188,13 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
taosMemoryFreeClear(pBlock->pBlockAgg); taosMemoryFreeClear(pBlock->pBlockAgg);
if (*status == FUNC_DATA_REQUIRED_FILTEROUT) { if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {
qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey, qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->filterOutBlocks += 1; pCost->filterOutBlocks += 1;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey, qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->skipBlocks += 1; pCost->skipBlocks += 1;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
@ -218,12 +217,12 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else { // failed to load the block sma data, data block statistics does not exist, load data block instead } else { // failed to load the block sma data, data block statistics does not exist, load data block instead
*status = FUNC_DATA_REQUIRED_DATA_LOAD; *status = FUNC_DATA_REQUIRED_DATA_LOAD;
} }
} }
ASSERT (*status == FUNC_DATA_REQUIRED_DATA_LOAD); ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);
// todo filter data block according to the block sma data firstly // todo filter data block according to the block sma data firstly
#if 0 #if 0
@ -249,8 +248,8 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
doFilter(pTableScanInfo->pFilterNode, pBlock); doFilter(pTableScanInfo->pFilterNode, pBlock);
if (pBlock->info.rows == 0) { if (pBlock->info.rows == 0) {
pCost->filterOutBlocks += 1; pCost->filterOutBlocks += 1;
qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey, qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -348,9 +347,9 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator, bool* newgroup) {
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN; pTableScanInfo->scanFlag = REPEAT_SCAN;
// if (pResultRowInfo->size > 0) { // if (pResultRowInfo->size > 0) {
// pResultRowInfo->curPos = 0; // pResultRowInfo->curPos = 0;
// } // }
qDebug("%s start to repeat scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64, qDebug("%s start to repeat scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey); GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
@ -367,7 +366,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator, bool* newgroup) {
GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey); GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
if (pResultRowInfo->size > 0) { if (pResultRowInfo->size > 0) {
// pResultRowInfo->curPos = pResultRowInfo->size - 1; // pResultRowInfo->curPos = pResultRowInfo->size - 1;
} }
p = doTableScanImpl(pOperator, newgroup); p = doTableScanImpl(pOperator, newgroup);
@ -376,9 +375,10 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator, bool* newgroup) {
return p; return p;
} }
SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t dataLoadFlag, SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput,
int32_t repeatTime, int32_t reverseTime, SArray* pColMatchInfo, SSDataBlock* pResBlock, int32_t dataLoadFlag, int32_t repeatTime, int32_t reverseTime,
SNode* pCondition, SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo) { SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition,
SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo) {
assert(repeatTime > 0); assert(repeatTime > 0);
STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo)); STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
@ -391,26 +391,26 @@ SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order,
return NULL; return NULL;
} }
pInfo->interval = *pInterval; pInfo->interval = *pInterval;
pInfo->sampleRatio = sampleRatio; pInfo->sampleRatio = sampleRatio;
pInfo->dataBlockLoadFlag= dataLoadFlag; pInfo->dataBlockLoadFlag = dataLoadFlag;
pInfo->pResBlock = pResBlock; pInfo->pResBlock = pResBlock;
pInfo->pFilterNode = pCondition; pInfo->pFilterNode = pCondition;
pInfo->dataReader = pTsdbReadHandle; pInfo->dataReader = pTsdbReadHandle;
pInfo->times = repeatTime; pInfo->times = repeatTime;
pInfo->reverseTimes = reverseTime; pInfo->reverseTimes = reverseTime;
pInfo->order = order; pInfo->order = order;
pInfo->current = 0; pInfo->current = 0;
pInfo->scanFlag = MAIN_SCAN; pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColMatchInfo; pInfo->pColMatchInfo = pColMatchInfo;
pOperator->name = "TableScanOperator"; pOperator->name = "TableScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
pOperator->blockingOptr = false; pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->numOfOutput = numOfOutput; pOperator->numOfOutput = numOfOutput;
pOperator->getNextFn = doTableScan; pOperator->getNextFn = doTableScan;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
static int32_t cost = 0; static int32_t cost = 0;
pOperator->cost.openCost = ++cost; pOperator->cost.openCost = ++cost;
@ -456,67 +456,67 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator, bool* newgroup) {
++numRowSteps; ++numRowSteps;
} }
tableBlockDist.dataBlockInfos = taosArrayInit(numRowSteps, sizeof(SFileBlockInfo)); tableBlockDist.dataBlockInfos = taosArrayInit(numRowSteps, sizeof(SFileBlockInfo));
taosArraySetSize(tableBlockDist.dataBlockInfos, numRowSteps); taosArraySetSize(tableBlockDist.dataBlockInfos, numRowSteps);
tableBlockDist.maxRows = INT_MIN; tableBlockDist.maxRows = INT_MIN;
tableBlockDist.minRows = INT_MAX; tableBlockDist.minRows = INT_MAX;
tsdbGetFileBlocksDistInfo(pTableScanInfo->dataReader, &tableBlockDist); tsdbGetFileBlocksDistInfo(pTableScanInfo->dataReader, &tableBlockDist);
tableBlockDist.numOfRowsInMemTable = (int32_t) tsdbGetNumOfRowsInMemTable(pTableScanInfo->dataReader); tableBlockDist.numOfRowsInMemTable = (int32_t)tsdbGetNumOfRowsInMemTable(pTableScanInfo->dataReader);
SSDataBlock* pBlock = pTableScanInfo->pResBlock; SSDataBlock* pBlock = pTableScanInfo->pResBlock;
pBlock->info.rows = 1; pBlock->info.rows = 1;
pBlock->info.numOfCols = 1; pBlock->info.numOfCols = 1;
// SBufferWriter bw = tbufInitWriter(NULL, false); // SBufferWriter bw = tbufInitWriter(NULL, false);
// blockDistInfoToBinary(&tableBlockDist, &bw); // blockDistInfoToBinary(&tableBlockDist, &bw);
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, 0); SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, 0);
// int32_t len = (int32_t) tbufTell(&bw); // int32_t len = (int32_t) tbufTell(&bw);
// pColInfo->pData = taosMemoryMalloc(len + sizeof(int32_t)); // pColInfo->pData = taosMemoryMalloc(len + sizeof(int32_t));
// *(int32_t*) pColInfo->pData = len; // *(int32_t*) pColInfo->pData = len;
// memcpy(pColInfo->pData + sizeof(int32_t), tbufGetData(&bw, false), len); // memcpy(pColInfo->pData + sizeof(int32_t), tbufGetData(&bw, false), len);
// //
// tbufCloseWriter(&bw); // tbufCloseWriter(&bw);
// SArray* g = GET_TABLEGROUP(pOperator->, 0); // SArray* g = GET_TABLEGROUP(pOperator->, 0);
// pOperator->pRuntimeEnv->current = taosArrayGetP(g, 0); // pOperator->pRuntimeEnv->current = taosArrayGetP(g, 0);
pOperator->status = OP_EXEC_DONE; pOperator->status = OP_EXEC_DONE;
return pBlock; return pBlock;
} }
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo) { SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo) {
STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo)); STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
goto _error; goto _error;
} }
pInfo->dataReader = dataReader; pInfo->dataReader = dataReader;
// pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData)); // pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0}; SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_BINARY; infoData.info.type = TSDB_DATA_TYPE_BINARY;
infoData.info.bytes = 1024; infoData.info.bytes = 1024;
infoData.info.colId = 0; infoData.info.colId = 0;
// taosArrayPush(pInfo->block.pDataBlock, &infoData); // taosArrayPush(pInfo->block.pDataBlock, &infoData);
pOperator->name = "DataBlockInfoScanOperator"; pOperator->name = "DataBlockInfoScanOperator";
// pOperator->operatorType = OP_TableBlockInfoScan; // pOperator->operatorType = OP_TableBlockInfoScan;
pOperator->blockingOptr = false; pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->_openFn = operatorDummyOpenFn; pOperator->_openFn = operatorDummyOpenFn;
pOperator->getNextFn = doBlockInfoScan; pOperator->getNextFn = doBlockInfoScan;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
return pOperator; return pOperator;
_error: _error:
taosMemoryFreeClear(pInfo); taosMemoryFreeClear(pInfo);
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
return NULL; return NULL;
@ -558,18 +558,18 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
blockDataCleanup(pInfo->pRes); blockDataCleanup(pInfo->pRes);
while (tqNextDataBlock(pInfo->readerHandle)) { while (tqNextDataBlock(pInfo->readerHandle)) {
pTaskInfo->code = tqRetrieveDataBlockInfo(pInfo->readerHandle, pBlockInfo); SArray* pCols = NULL;
if (pTaskInfo->code != TSDB_CODE_SUCCESS) { uint64_t groupId;
terrno = pTaskInfo->code; int32_t numOfRows;
pOperator->status = OP_EXEC_DONE; int32_t code = tqRetrieveDataBlock(&pCols, pInfo->readerHandle, &groupId, &numOfRows);
if (code != TSDB_CODE_SUCCESS || numOfRows == 0) {
pTaskInfo->code = code;
return NULL; return NULL;
} }
if (pBlockInfo->rows == 0) { pInfo->pRes->info.groupId = groupId;
break; pInfo->pRes->info.rows = numOfRows;
}
SArray* pCols = tqRetrieveDataBlock(pInfo->readerHandle);
int32_t numOfCols = pInfo->pRes->info.numOfCols; int32_t numOfCols = pInfo->pRes->info.numOfCols;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
@ -579,7 +579,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
} }
bool colExists = false; bool colExists = false;
for(int32_t j = 0; j < taosArrayGetSize(pCols); ++j) { for (int32_t j = 0; j < taosArrayGetSize(pCols); ++j) {
SColumnInfoData* pResCol = taosArrayGet(pCols, j); SColumnInfoData* pResCol = taosArrayGet(pCols, j);
if (pResCol->info.colId == pColMatchInfo->colId) { if (pResCol->info.colId == pColMatchInfo->colId) {
taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol); taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol);
@ -618,7 +618,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator, bool* newgroup)
} }
} }
SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo) { SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
SArray* pTableIdList, SExecTaskInfo* pTaskInfo) {
SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo)); SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -631,7 +632,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
int32_t numOfOutput = taosArrayGetSize(pColList); int32_t numOfOutput = taosArrayGetSize(pColList);
SArray* pColIds = taosArrayInit(4, sizeof(int16_t)); SArray* pColIds = taosArrayInit(4, sizeof(int16_t));
for(int32_t i = 0; i < numOfOutput; ++i) { for (int32_t i = 0; i < numOfOutput; ++i) {
int16_t* id = taosArrayGet(pColList, i); int16_t* id = taosArrayGet(pColList, i);
taosArrayPush(pColIds, id); taosArrayPush(pColIds, id);
} }
@ -657,16 +658,16 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
pInfo->readerHandle = streamReadHandle; pInfo->readerHandle = streamReadHandle;
pInfo->pRes = pResBlock; pInfo->pRes = pResBlock;
pOperator->name = "StreamBlockScanOperator"; pOperator->name = "StreamBlockScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
pOperator->blockingOptr = false; pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols; pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->_openFn = operatorDummyOpenFn; pOperator->_openFn = operatorDummyOpenFn;
pOperator->getNextFn = doStreamBlockScan; pOperator->getNextFn = doStreamBlockScan;
pOperator->closeFn = operatorDummyCloseFn; pOperator->closeFn = operatorDummyCloseFn;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
return pOperator; return pOperator;
} }
@ -746,9 +747,9 @@ static int32_t loadSysTableContentCb(void* param, const SDataBuf* pMsg, int32_t
SRetrieveMetaTableRsp* pRsp = pScanResInfo->pRsp; SRetrieveMetaTableRsp* pRsp = pScanResInfo->pRsp;
pRsp->numOfRows = htonl(pRsp->numOfRows); pRsp->numOfRows = htonl(pRsp->numOfRows);
pRsp->useconds = htobe64(pRsp->useconds); pRsp->useconds = htobe64(pRsp->useconds);
pRsp->handle = htobe64(pRsp->handle); pRsp->handle = htobe64(pRsp->handle);
pRsp->compLen = htonl(pRsp->compLen); pRsp->compLen = htonl(pRsp->compLen);
} else { } else {
operator->pTaskInfo->code = code; operator->pTaskInfo->code = code;
} }
@ -790,7 +791,7 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
if (rowRes[j] == 0) { if (rowRes[j] == 0) {
continue; continue;
} }
colDataAppend(pDest, numOfRow, colDataGetData(pSrc, j), false); colDataAppend(pDest, numOfRow, colDataGetData(pSrc, j), false);
numOfRow += 1; numOfRow += 1;
} }
@ -841,7 +842,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator, bool* newgroup) {
SColumnInfoData* pColInfoData = taosArrayGet(pInfo->pRes->pDataBlock, i); SColumnInfoData* pColInfoData = taosArrayGet(pInfo->pRes->pDataBlock, i);
int64_t tmp = 0; int64_t tmp = 0;
char t[10] = {0}; char t[10] = {0};
STR_TO_VARSTR(t, "_"); //TODO STR_TO_VARSTR(t, "_"); // TODO
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
colDataAppend(pColInfoData, numOfRows, t, false); colDataAppend(pColInfoData, numOfRows, t, false);
} else { } else {
@ -939,12 +940,12 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
return NULL; return NULL;
} }
pInfo->accountId = accountId; pInfo->accountId = accountId;
pInfo->showRewrite = showRewrite; pInfo->showRewrite = showRewrite;
pInfo->pRes = pResBlock; pInfo->pRes = pResBlock;
pInfo->capacity = 4096; pInfo->capacity = 4096;
pInfo->pCondition = pCondition; pInfo->pCondition = pCondition;
pInfo->scanCols = colList; pInfo->scanCols = colList;
// TODO remove it // TODO remove it
int32_t tableType = 0; int32_t tableType = 0;
@ -999,9 +1000,9 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
tableType = TSDB_MGMT_TABLE_CONNS; tableType = TSDB_MGMT_TABLE_CONNS;
} else if (strncasecmp(name, TSDB_INS_TABLE_QUERIES, tListLen(pName->tname)) == 0) { } else if (strncasecmp(name, TSDB_INS_TABLE_QUERIES, tListLen(pName->tname)) == 0) {
tableType = TSDB_MGMT_TABLE_QUERIES; tableType = TSDB_MGMT_TABLE_QUERIES;
} else if (strncasecmp(name, TSDB_INS_TABLE_VNODES, tListLen(pName->tname)) == 0) { } else if (strncasecmp(name, TSDB_INS_TABLE_VNODES, tListLen(pName->tname)) == 0) {
tableType = TSDB_MGMT_TABLE_VNODES; tableType = TSDB_MGMT_TABLE_VNODES;
}else { } else {
ASSERT(0); ASSERT(0);
} }
@ -1038,15 +1039,15 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
#endif #endif
} }
pOperator->name = "SysTableScanOperator"; pOperator->name = "SysTableScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN;
pOperator->blockingOptr = false; pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols; pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->getNextFn = doSysTableScan; pOperator->getNextFn = doSysTableScan;
pOperator->closeFn = destroySysScanOperator; pOperator->closeFn = destroySysScanOperator;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
return pOperator; return pOperator;
} }