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

[td-2859]refactor.

This commit is contained in:
Haojun Liao 2021-04-08 11:27:38 +08:00
parent 4b86150bb6
commit 77dfb9fe28
16 changed files with 715 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/inc/tscLocalMerge.h
View File

@ -62,6 +62,8 @@ typedef struct SLocalMerger {
bool discard;
int32_t offset; // limit offset value
bool orderPrjOnSTable; // projection query on stable
char* tagBuf; // max tag buffer
int32_t tagBufLen;
} SLocalMerger;
typedef struct SRetrieveSupport {

5
src/client/inc/tscUtil.h
View File

@ -98,10 +98,9 @@ typedef struct SMergeTsCtx {
int8_t compared;
}SMergeTsCtx;
typedef struct SVgroupTableInfo {
SVgroupInfo vgInfo;
SArray* itemList; //SArray<STableIdInfo>
SArray *itemList; // SArray<STableIdInfo>
} SVgroupTableInfo;
static FORCE_INLINE SQueryInfo* tscGetQueryInfo(SSqlCmd* pCmd, int32_t subClauseIndex) {
@ -321,7 +320,7 @@ STableMeta* tscTableMetaDup(STableMeta* pTableMeta);
int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr);
void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx, SSchema* pSchema);
void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr);
void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr, int32_t stage);
void* malloc_throw(size_t size);
void* calloc_throw(size_t nmemb, size_t size);

4
src/client/inc/tsclient.h
View File

@ -213,7 +213,7 @@ typedef struct SQueryInfo {
int32_t round; // 0/1/....
int32_t bufLen;
char* buf;
SQInfo* pQInfo; // global merge operator
SArray* pDSOperator; // data source operator
SArray* pPhyOperator; // physical query execution plan
SQueryAttr* pQueryAttr; // query object
@ -420,6 +420,8 @@ void tscRestoreFuncForSTableQuery(SQueryInfo *pQueryInfo);
int32_t tscCreateResPointerInfo(SSqlRes *pRes, SQueryInfo *pQueryInfo);
void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo);
void tscSetResRawPtrRv(SSqlRes* pRes, SQueryInfo* pQueryInfo, SSDataBlock* pBlock);
void prepareInputDataFromUpstream(SSqlRes* pRes, SQueryInfo* pQueryInfo);
void tscResetSqlCmd(SSqlCmd *pCmd, bool removeMeta);

430
src/client/src/tscLocalMerge.c
View File

@ -30,6 +30,8 @@ typedef struct SCompareParam {
int32_t groupOrderType;
} SCompareParam;
bool needToMergeRv(SSDataBlock* pBlock, SLocalMerger *pLocalMerge, int32_t index, char **buf);
int32_t treeComparator(const void *pLeft, const void *pRight, void *param) {
int32_t pLeftIdx = *(int32_t *)pLeft;
int32_t pRightIdx = *(int32_t *)pRight;
@ -88,6 +90,7 @@ void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx, SSchem
pCtx[i].param[2].i64 = pQueryInfo->order.order;
pCtx[i].param[2].nType = TSDB_DATA_TYPE_BIGINT;
pCtx[i].param[1].i64 = pQueryInfo->order.orderColId;
pCtx[i].param[0].i64 = pExpr->base.param[0].i64; // top/bot parameter
} else if (functionId == TSDB_FUNC_APERCT) {
pCtx[i].param[0].i64 = pExpr->base.param[0].i64;
pCtx[i].param[0].nType = pExpr->base.param[0].nType;
@ -126,17 +129,17 @@ void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx, SSchem
}
}
static void setCtxInputOutputBuffer(SQueryInfo* pQueryInfo, SQLFunctionCtx *pCtx, SLocalMerger *pReducer,
tOrderDescriptor *pDesc) {
static UNUSED_FUNC void setCtxInputOutputBuffer(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx, SLocalMerger *pMerger,
tOrderDescriptor *pDesc) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
SExprInfo *pExpr = tscSqlExprGet(pQueryInfo, i);
pCtx[i].pOutput = pReducer->pResultBuf->data + pExpr->base.offset * pReducer->resColModel->capacity;
pCtx[i].pOutput = pMerger->pResultBuf->data + pExpr->base.offset * pMerger->resColModel->capacity;
// input buffer hold only one point data
int16_t offset = getColumnModelOffset(pDesc->pColumnModel, i);
pCtx[i].pInput = pReducer->pTempBuffer->data + offset;
pCtx[i].pInput = pMerger->pTempBuffer->data + offset;
int32_t functionId = pCtx[i].functionId;
if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM || functionId == TSDB_FUNC_DIFF) {
@ -227,8 +230,8 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
size_t size = sizeof(SLocalMerger) + POINTER_BYTES * numOfFlush;
SLocalMerger *pReducer = (SLocalMerger *) calloc(1, size);
if (pReducer == NULL) {
SLocalMerger *pMerger = (SLocalMerger *) calloc(1, size);
if (pMerger == NULL) {
tscError("%p failed to create local merge structure, out of memory", pSql);
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, finalmodel, pFFModel, numOfBuffer);
@ -236,15 +239,15 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
return;
}
pReducer->pExtMemBuffer = pMemBuffer;
pReducer->pLocalDataSrc = (SLocalDataSource **)&pReducer[1];
assert(pReducer->pLocalDataSrc != NULL);
pMerger->pExtMemBuffer = pMemBuffer;
pMerger->pLocalDataSrc = (SLocalDataSource **)&pMerger[1];
assert(pMerger->pLocalDataSrc != NULL);
pReducer->numOfBuffer = numOfFlush;
pReducer->numOfVnode = numOfBuffer;
pMerger->numOfBuffer = numOfFlush;
pMerger->numOfVnode = numOfBuffer;
pReducer->pDesc = pDesc;
tscDebug("%p the number of merged leaves is: %d", pSql, pReducer->numOfBuffer);
pMerger->pDesc = pDesc;
tscDebug("%p the number of merged leaves is: %d", pSql, pMerger->numOfBuffer);
int32_t idx = 0;
for (int32_t i = 0; i < numOfBuffer; ++i) {
@ -255,11 +258,11 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
if (ds == NULL) {
tscError("%p failed to create merge structure", pSql);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tfree(pReducer);
tfree(pMerger);
return;
}
pReducer->pLocalDataSrc[idx] = ds;
pMerger->pLocalDataSrc[idx] = ds;
ds->pMemBuffer = pMemBuffer[i];
ds->flushoutIdx = j;
@ -292,84 +295,84 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
// no data actually, no need to merge result.
if (idx == 0) {
tfree(pReducer);
tfree(pMerger);
return;
}
pReducer->numOfBuffer = idx;
pMerger->numOfBuffer = idx;
SCompareParam *param = malloc(sizeof(SCompareParam));
if (param == NULL) {
tfree(pReducer);
tfree(pMerger);
return;
}
param->pLocalData = pReducer->pLocalDataSrc;
param->pDesc = pReducer->pDesc;
param->num = pReducer->pLocalDataSrc[0]->pMemBuffer->numOfElemsPerPage;
param->pLocalData = pMerger->pLocalDataSrc;
param->pDesc = pMerger->pDesc;
param->num = pMerger->pLocalDataSrc[0]->pMemBuffer->numOfElemsPerPage;
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
param->groupOrderType = pQueryInfo->groupbyExpr.orderType;
pReducer->orderPrjOnSTable = tscOrderedProjectionQueryOnSTable(pQueryInfo, 0);
pMerger->orderPrjOnSTable = tscOrderedProjectionQueryOnSTable(pQueryInfo, 0);
pRes->code = tLoserTreeCreate(&pReducer->pLoserTree, pReducer->numOfBuffer, param, treeComparator);
if (pReducer->pLoserTree == NULL || pRes->code != 0) {
pRes->code = tLoserTreeCreate(&pMerger->pLoserTree, pMerger->numOfBuffer, param, treeComparator);
if (pMerger->pLoserTree == NULL || pRes->code != 0) {
tfree(param);
tfree(pReducer);
tfree(pMerger);
return;
}
// the input data format follows the old format, but output in a new format.
// so, all the input must be parsed as old format
pReducer->pCtx = (SQLFunctionCtx *)calloc(tscSqlExprNumOfExprs(pQueryInfo), sizeof(SQLFunctionCtx));
pReducer->rowSize = pMemBuffer[0]->nElemSize;
pMerger->pCtx = (SQLFunctionCtx *)calloc(tscSqlExprNumOfExprs(pQueryInfo), sizeof(SQLFunctionCtx));
pMerger->rowSize = pMemBuffer[0]->nElemSize;
tscRestoreFuncForSTableQuery(pQueryInfo);
tscFieldInfoUpdateOffset(pQueryInfo);
if (pReducer->rowSize > pMemBuffer[0]->pageSize) {
if (pMerger->rowSize > pMemBuffer[0]->pageSize) {
assert(false); // todo fixed row size is larger than the minimum page size;
}
pReducer->hasPrevRow = false;
pReducer->hasUnprocessedRow = false;
pMerger->hasPrevRow = false;
pMerger->hasUnprocessedRow = false;
pReducer->prevRowOfInput = (char *)calloc(1, pReducer->rowSize);
pMerger->prevRowOfInput = (char *)calloc(1, pMerger->rowSize);
// used to keep the latest input row
pReducer->pTempBuffer = (tFilePage *)calloc(1, pReducer->rowSize + sizeof(tFilePage));
pReducer->discardData = (tFilePage *)calloc(1, pReducer->rowSize + sizeof(tFilePage));
pReducer->discard = false;
pMerger->pTempBuffer = (tFilePage *)calloc(1, pMerger->rowSize + sizeof(tFilePage));
pMerger->discardData = (tFilePage *)calloc(1, pMerger->rowSize + sizeof(tFilePage));
pMerger->discard = false;
pReducer->nResultBufSize = pMemBuffer[0]->pageSize * 16;
pReducer->pResultBuf = (tFilePage *)calloc(1, pReducer->nResultBufSize + sizeof(tFilePage));
pMerger->nResultBufSize = pMemBuffer[0]->pageSize * 16;
pMerger->pResultBuf = (tFilePage *)calloc(1, pMerger->nResultBufSize + sizeof(tFilePage));
pReducer->resColModel = finalmodel;
pReducer->resColModel->capacity = pReducer->nResultBufSize;
pReducer->finalModel = pFFModel;
pMerger->resColModel = finalmodel;
pMerger->resColModel->capacity = pMerger->nResultBufSize;
pMerger->finalModel = pFFModel;
if (finalmodel->rowSize > 0) {
pReducer->resColModel->capacity /= finalmodel->rowSize;
pMerger->resColModel->capacity /= finalmodel->rowSize;
}
assert(finalmodel->rowSize > 0 && finalmodel->rowSize <= pReducer->rowSize);
pReducer->pFinalRes = calloc(1, pReducer->rowSize * pReducer->resColModel->capacity);
assert(finalmodel->rowSize > 0 && finalmodel->rowSize <= pMerger->rowSize);
pMerger->pFinalRes = calloc(1, pMerger->rowSize * pMerger->resColModel->capacity);
if (pReducer->pTempBuffer == NULL || pReducer->discardData == NULL || pReducer->pResultBuf == NULL ||
pReducer->pFinalRes == NULL || pReducer->prevRowOfInput == NULL) {
tfree(pReducer->pTempBuffer);
tfree(pReducer->discardData);
tfree(pReducer->pResultBuf);
tfree(pReducer->pFinalRes);
tfree(pReducer->prevRowOfInput);
tfree(pReducer->pLoserTree);
if (pMerger->pTempBuffer == NULL || pMerger->discardData == NULL || pMerger->pResultBuf == NULL ||
pMerger->pFinalRes == NULL || pMerger->prevRowOfInput == NULL) {
tfree(pMerger->pTempBuffer);
tfree(pMerger->discardData);
tfree(pMerger->pResultBuf);
tfree(pMerger->pFinalRes);
tfree(pMerger->prevRowOfInput);
tfree(pMerger->pLoserTree);
tfree(param);
tfree(pReducer);
tfree(pMerger);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
return;
}
pReducer->pTempBuffer->num = 0;
pMerger->pTempBuffer->num = 0;
tscCreateResPointerInfo(pRes, pQueryInfo);
@ -378,13 +381,23 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
pschema[i] = pDesc->pColumnModel->pFields[i].field;
}
tsCreateSQLFunctionCtx(pQueryInfo, pReducer->pCtx, pschema);
setCtxInputOutputBuffer(pQueryInfo, pReducer->pCtx, pReducer, pDesc);
tsCreateSQLFunctionCtx(pQueryInfo, pMerger->pCtx, pschema);
// setCtxInputOutputBuffer(pQueryInfo, pMerger->pCtx, pMerger, pDesc);
tfree(pschema);
int32_t maxBufSize = 0;
for (int32_t k = 0; k < tscSqlExprNumOfExprs(pQueryInfo); ++k) {
SExprInfo *pExpr = tscSqlExprGet(pQueryInfo, k);
if (maxBufSize < pExpr->base.resBytes && pExpr->base.functionId == TSDB_FUNC_TAG) {
maxBufSize = pExpr->base.resBytes;
}
}
pMerger->tagBuf = calloc(1, maxBufSize);
// we change the capacity of schema to denote that there is only one row in temp buffer
pReducer->pDesc->pColumnModel->capacity = 1;
pMerger->pDesc->pColumnModel->capacity = 1;
// restore the limitation value at the last stage
if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
@ -392,9 +405,9 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
pQueryInfo->limit.offset = pQueryInfo->prjOffset;
}
pReducer->offset = (int32_t)pQueryInfo->limit.offset;
pMerger->offset = (int32_t)pQueryInfo->limit.offset;
pRes->pLocalMerger = pReducer;
pRes->pLocalMerger = pMerger;
pRes->numOfGroups = 0;
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
@ -405,9 +418,10 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
if (pQueryInfo->fillType != TSDB_FILL_NONE) {
SFillColInfo* pFillCol = createFillColInfo(pQueryInfo);
pReducer->pFillInfo = taosCreateFillInfo(pQueryInfo->order.order, revisedSTime, pQueryInfo->groupbyExpr.numOfGroupCols,
4096, (int32_t)pQueryInfo->fieldsInfo.numOfOutput, pQueryInfo->interval.sliding, pQueryInfo->interval.slidingUnit,
tinfo.precision, pQueryInfo->fillType, pFillCol, pSql);
pMerger->pFillInfo =
taosCreateFillInfo(pQueryInfo->order.order, revisedSTime, pQueryInfo->groupbyExpr.numOfGroupCols, 4096,
(int32_t)pQueryInfo->fieldsInfo.numOfOutput, pQueryInfo->interval.sliding,
pQueryInfo->interval.slidingUnit, tinfo.precision, pQueryInfo->fillType, pFillCol, pSql);
}
}
@ -626,12 +640,12 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SSqlCmd *pCm
}
}
bool isSameGroup(SSqlCmd *pCmd, SLocalMerger *pReducer, char *pPrev, tFilePage *tmpBuffer) {
bool isSameGroup(SSqlCmd *pCmd, SLocalMerger *pMerger, char *pPrev, tFilePage *tmpBuffer) {
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
// disable merge procedure for column projection query
int16_t functionId = pReducer->pCtx[0].functionId;
if (pReducer->orderPrjOnSTable) {
int16_t functionId = pMerger->pCtx[0].functionId;
if (pMerger->orderPrjOnSTable) {
return true;
}
@ -639,7 +653,7 @@ bool isSameGroup(SSqlCmd *pCmd, SLocalMerger *pReducer, char *pPrev, tFilePage *
return false;
}
tOrderDescriptor *pOrderDesc = pReducer->pDesc;
tOrderDescriptor *pOrderDesc = pMerger->pDesc;
SColumnOrderInfo* orderInfo = &pOrderDesc->orderInfo;
// no group by columns, all data belongs to one group
@ -1065,12 +1079,9 @@ static void savePreviousRow(SLocalMerger *pLocalMerge, tFilePage *tmpBuffer) {
pLocalMerge->hasPrevRow = true;
}
static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool needInit) {
static void doExecuteFinalMerge( SLocalMerger *pLocalMerge, int32_t numOfExpr, bool needInit) {
// the tag columns need to be set before all functions execution
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t j = 0; j < size; ++j) {
for (int32_t j = 0; j < numOfExpr; ++j) {
SQLFunctionCtx *pCtx = &pLocalMerge->pCtx[j];
// tags/tags_dummy function, the tag field of SQLFunctionCtx is from the input buffer
@ -1087,8 +1098,8 @@ static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool n
}
} else if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM) {
SExprInfo *pExpr = tscSqlExprGet(pQueryInfo, j);
pCtx->param[0].i64 = pExpr->base.param[0].i64;
// SExprInfo *pExpr = tscSqlExprGet(pQueryInfo, j); // TODO this data is from
// pCtx->param[0].i64 = pExpr->base.param[0].i64;
}
pCtx->currentStage = MERGE_STAGE;
@ -1098,7 +1109,7 @@ static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool n
}
}
for (int32_t j = 0; j < size; ++j) {
for (int32_t j = 0; j < numOfExpr; ++j) {
int32_t functionId = pLocalMerge->pCtx[j].functionId;
if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
@ -1108,19 +1119,79 @@ static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool n
}
}
static void savePrevOrderColumns(SMultiwayMergeInfo* pInfo, SSDataBlock* pBlock, int32_t rowIndex) {
int32_t size = taosArrayGetSize(pInfo->orderColumnList);
for(int32_t i = 0; i < size; ++i) {
int32_t index = *(int16_t*)taosArrayGet(pInfo->orderColumnList, i);
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, index);
memcpy(pInfo->prevRow[i], pColInfo->pData + pColInfo->info.bytes * rowIndex, pColInfo->info.bytes);
}
}
static void doExecuteFinalMergeRv(SMultiwayMergeInfo* pInfo, int32_t numOfExpr, SSDataBlock* pBlock, bool needInit) {
SQLFunctionCtx* pCtx = pInfo->binfo.pCtx;
for(int32_t i = 0; i < pBlock->info.rows; ++i) {
if (pInfo->hasPrev) {
if (needToMergeRv(pBlock, pInfo->pMerge, i, pInfo->prevRow)) {
for (int32_t j = 0; j < numOfExpr; ++j) {
int32_t functionId = pCtx[j].functionId;
if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
aAggs[functionId].mergeFunc(&pCtx[j]);
}
} else {
// todo finalize the result
for (int32_t j = 0; j < numOfExpr; ++j) {
int32_t functionId = pCtx[j].functionId;
if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
pCtx[j].pOutput += pCtx[j].outputBytes;
pCtx[j].pInput += pCtx[j].inputBytes;
aAggs[functionId].mergeFunc(&pCtx[j]);
}
pInfo->binfo.pRes->info.rows += 1;
}
} else {
for (int32_t j = 0; j < numOfExpr; ++j) {
int32_t functionId = pCtx[j].functionId;
if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
aAggs[functionId].mergeFunc(&pCtx[j]);
}
}
savePrevOrderColumns(pInfo, pBlock, i);
pInfo->hasPrev = true;
}
}
static void handleUnprocessedRow(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, tFilePage *tmpBuffer) {
if (pLocalMerge->hasUnprocessedRow) {
pLocalMerge->hasUnprocessedRow = false;
doExecuteFinalMerge(pCmd, pLocalMerge, true);
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
doExecuteFinalMerge(pLocalMerge, size, true);
savePreviousRow(pLocalMerge, tmpBuffer);
}
}
static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx) {
static int64_t getNumOfResultLocal(SQLFunctionCtx *pCtx, int32_t numOfExprs) {
int64_t maxOutput = 0;
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t j = 0; j < size; ++j) {
for (int32_t j = 0; j < numOfExprs; ++j) {
/*
* ts, tag, tagprj function can not decide the output number of current query
* the number of output result is decided by main output
@ -1145,53 +1216,36 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx)
* filled with the same result, which is the tags, specified in group by clause
*
*/
static void fillMultiRowsOfTagsVal(SQueryInfo *pQueryInfo, int32_t numOfRes, SLocalMerger *pLocalMerge) {
int32_t maxBufSize = 0; // find the max tags column length to prepare the buffer
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < size; ++k) {
SExprInfo *pExpr = tscSqlExprGet(pQueryInfo, k);
if (maxBufSize < pExpr->base.resBytes && pExpr->base.functionId == TSDB_FUNC_TAG) {
maxBufSize = pExpr->base.resBytes;
}
}
assert(maxBufSize >= 0);
char *buf = malloc((size_t)maxBufSize);
for (int32_t k = 0; k < size; ++k) {
static void fillMultiRowsOfTagsVal(SLocalMerger *pLocalMerge, int32_t numOfRes, int32_t numOfExprs) {
for (int32_t k = 0; k < numOfExprs; ++k) {
SQLFunctionCtx *pCtx = &pLocalMerge->pCtx[k];
if (pCtx->functionId != TSDB_FUNC_TAG) {
continue;
}
int32_t inc = numOfRes - 1; // tsdb_func_tag function only produce one row of result
memset(buf, 0, (size_t)maxBufSize);
memcpy(buf, pCtx->pOutput, (size_t)pCtx->outputBytes);
memset(pLocalMerge->tagBuf, 0, (size_t)pLocalMerge->tagBufLen);
memcpy(pLocalMerge->tagBuf, pCtx->pOutput, (size_t)pCtx->outputBytes);
for (int32_t i = 0; i < inc; ++i) {
pCtx->pOutput += pCtx->outputBytes;
memcpy(pCtx->pOutput, buf, (size_t)pCtx->outputBytes);
memcpy(pCtx->pOutput, pLocalMerge->tagBuf, (size_t)pCtx->outputBytes);
}
}
free(buf);
}
int32_t finalizeRes(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < size; ++k) {
int32_t finalizeRes(SLocalMerger *pLocalMerge, int32_t numOfExprs) {
for (int32_t k = 0; k < numOfExprs; ++k) {
SQLFunctionCtx* pCtx = &pLocalMerge->pCtx[k];
aAggs[pCtx->functionId].xFinalize(pCtx);
}
pLocalMerge->hasPrevRow = false;
int32_t numOfRes = (int32_t)getNumOfResultLocal(pQueryInfo, pLocalMerge->pCtx);
int32_t numOfRes = (int32_t)getNumOfResultLocal(pLocalMerge->pCtx, numOfExprs);
pLocalMerge->pResultBuf->num += numOfRes;
fillMultiRowsOfTagsVal(pQueryInfo, numOfRes, pLocalMerge);
fillMultiRowsOfTagsVal(pLocalMerge, numOfRes, numOfExprs);
return numOfRes;
}
@ -1226,6 +1280,22 @@ bool needToMerge(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge, tFilePage *t
return (ret == 0);
}
bool needToMergeRv(SSDataBlock* pBlock, SLocalMerger *pLocalMerge, int32_t index, char **buf) {
int32_t ret = 0;
tOrderDescriptor *pDesc = pLocalMerge->pDesc;
if (pDesc->orderInfo.numOfCols > 0) {
if (pDesc->tsOrder == TSDB_ORDER_ASC) { // asc
// todo refactor comparator
ret = compare_aRv(pBlock, pDesc->orderInfo.colIndex, pDesc->orderInfo.numOfCols, index, buf, TSDB_ORDER_ASC);
} else { // desc
// ret = compare_d(pLocalMerge->pDesc, 1, 0, pLocalMerge->prevRowOfInput, 1, 0, tmpBuffer->data);
}
}
/* if ret == 0, means the result belongs to the same group */
return (ret == 0);
}
static bool reachGroupResultLimit(SQueryInfo *pQueryInfo, SSqlRes *pRes) {
return (pRes->numOfGroups >= pQueryInfo->slimit.limit && pQueryInfo->slimit.limit >= 0);
}
@ -1310,6 +1380,24 @@ bool genFinalResults(SSqlObj *pSql, SLocalMerger *pLocalMerge, bool noMoreCurren
return true;
}
bool genFinalResultsRv(SSqlObj *pSql, SLocalMerger *pLocalMerge, bool noMoreCurrentGroupRes) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
SQueryInfo * pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
tFilePage * pResBuf = pLocalMerge->pResultBuf;
SColumnModel *pModel = pLocalMerge->resColModel;
pRes->code = TSDB_CODE_SUCCESS;
tColModelCompact(pModel, pResBuf, pModel->capacity);
// no interval query, no fill operation
genFinalResWithoutFill(pRes, pLocalMerge, pQueryInfo);
return true;
}
void resetOutputBuf(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge) {// reset output buffer to the beginning
size_t t = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < t; ++i) {
@ -1437,7 +1525,7 @@ static void doProcessResultInNextWindow(SSqlObj *pSql, int32_t numOfRes) {
}
}
doExecuteFinalMerge(pCmd, pLocalMerge, true);
doExecuteFinalMerge(pLocalMerge, size, true);
}
int32_t tscDoLocalMerge(SSqlObj *pSql) {
@ -1445,8 +1533,9 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
tscResetForNextRetrieve(pRes);
assert(pSql->signature == pSql);
if (pSql->signature != pSql || pRes == NULL || pRes->pLocalMerger == NULL) { // all data has been processed
if (pRes->pLocalMerger == NULL) { // all data has been processed
if (pRes->code == TSDB_CODE_SUCCESS) {
return pRes->code;
}
@ -1459,6 +1548,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
tFilePage *tmpBuffer = pLocalMerge->pTempBuffer;
int32_t numOfExprs = (int32_t) tscSqlExprNumOfExprs(pQueryInfo);
if (doHandleLastRemainData(pSql)) {
return TSDB_CODE_SUCCESS;
}
@ -1481,6 +1571,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
#ifdef _DEBUG_VIEW
printf("chosen data in pTree[0] = %d\n", pTree->pNode[0].index);
#endif
assert((pTree->pNode[0].index < pLocalMerge->numOfBuffer) && (pTree->pNode[0].index >= 0) && tmpBuffer->num == 0);
// chosen from loser tree
@ -1529,7 +1620,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
if (pLocalMerge->hasPrevRow) {
if (needToMerge(pQueryInfo, pLocalMerge, tmpBuffer)) {
// belong to the group of the previous row, continue process it
doExecuteFinalMerge(pCmd, pLocalMerge, false);
doExecuteFinalMerge(pLocalMerge, numOfExprs, false);
// copy to buffer
savePreviousRow(pLocalMerge, tmpBuffer);
@ -1538,7 +1629,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
* current row does not belong to the group of previous row.
* so the processing of previous group is completed.
*/
int32_t numOfRes = finalizeRes(pQueryInfo, pLocalMerge);
int32_t numOfRes = finalizeRes(pLocalMerge, numOfExprs);
bool sameGroup = isSameGroup(pCmd, pLocalMerge, pLocalMerge->prevRowOfInput, tmpBuffer);
tFilePage *pResBuf = pLocalMerge->pResultBuf;
@ -1601,7 +1692,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
}
}
} else {
doExecuteFinalMerge(pCmd, pLocalMerge, true);
doExecuteFinalMerge(pLocalMerge, numOfExprs,true);
savePreviousRow(pLocalMerge, tmpBuffer); // copy the processed row to buffer
}
@ -1610,7 +1701,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) {
}
if (pLocalMerge->hasPrevRow) {
finalizeRes(pQueryInfo, pLocalMerge);
finalizeRes(pLocalMerge, numOfExprs);
}
if (pLocalMerge->pResultBuf->num) {
@ -1686,3 +1777,126 @@ int32_t doArithmeticCalculate(SQueryInfo* pQueryInfo, tFilePage* pOutput, int32_
return offset;
}
#define COLMODEL_GET_VAL(data, schema, allrow, rowId, colId) \
(data + (schema)->pFields[colId].offset * (allrow) + (rowId) * (schema)->pFields[colId].field.bytes)
static void appendOneRowToDataBlock(SSDataBlock *pBlock, char *buf, SColumnModel *pModel, int32_t rowIndex,
int32_t maxRows) {
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
char* p = pColInfo->pData + pBlock->info.rows * pColInfo->info.bytes;
// char *dst = COLMODEL_GET_VAL(dstPage->data, dstModel, dstModel->capacity, dstPage->num, col);
char *src = COLMODEL_GET_VAL(buf, pModel, maxRows, rowIndex, i);
// char* src = buf + rowIndex * pColInfo->info.bytes;
memmove(p, src, pColInfo->info.bytes);
}
pBlock->info.rows += 1;
}
static SSDataBlock* doMultiwaySort(void* param) {
SOperatorInfo* pOperator = (SOperatorInfo*) param;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
SMultiwayMergeInfo *pInfo = pOperator->info;
SLocalMerger *pMerger = pInfo->pMerge;
SLoserTreeInfo *pTree = pMerger->pLoserTree;
SColumnModel *pModel = pMerger->pDesc->pColumnModel;
tFilePage *tmpBuffer = pMerger->pTempBuffer;
pInfo->binfo.pRes->info.rows = 0;
while(1) {
if (isAllSourcesCompleted(pMerger)) {
break;
}
#ifdef _DEBUG_VIEW
printf("chosen data in pTree[0] = %d\n", pTree->pNode[0].index);
#endif
assert((pTree->pNode[0].index < pMerger->numOfBuffer) && (pTree->pNode[0].index >= 0) && tmpBuffer->num == 0);
// chosen from loser tree
SLocalDataSource *pOneDataSrc = pMerger->pLocalDataSrc[pTree->pNode[0].index];
appendOneRowToDataBlock(pInfo->binfo.pRes, pOneDataSrc->filePage.data, pModel, pOneDataSrc->rowIdx, pOneDataSrc->pMemBuffer->pColumnModel->capacity);
#if defined(_DEBUG_VIEW)
printf("chosen row:\t");
SSrcColumnInfo colInfo[256] = {0};
tscGetSrcColumnInfo(colInfo, pQueryInfo);
tColModelDisplayEx(pModel, tmpBuffer->data, tmpBuffer->num, pModel->capacity, colInfo);
#endif
pOneDataSrc->rowIdx += 1;
adjustLoserTreeFromNewData(pMerger, pOneDataSrc, pTree);
if (pInfo->binfo.pRes->info.rows >= 4096) { // TODO threshold
return pInfo->binfo.pRes;
}
}
return (pInfo->binfo.pRes->info.rows > 0)? pInfo->binfo.pRes:NULL;
}
SOperatorInfo *createMultiwaySortOperatorInfo(SQueryRuntimeEnv *pRuntimeEnv, SExprInfo *pExpr, int32_t numOfOutput,
int32_t numOfRows, void *merger) {
SMultiwayMergeInfo* pInfo = calloc(1, sizeof(SMultiwayMergeInfo));
pInfo->pMerge = merger;
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, numOfRows);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
pOperator->name = "MultiwaySortOperator";
pOperator->operatorType = OP_MultiwaySort;
pOperator->blockingOptr = false;
pOperator->status = OP_IN_EXECUTING;
pOperator->info = pInfo;
pOperator->pRuntimeEnv = pRuntimeEnv;
pOperator->numOfOutput = pRuntimeEnv->pQueryAttr->numOfCols;
pOperator->exec = doMultiwaySort;
return pOperator;
}
SSDataBlock* doGlobalAggregate(void* param) {
SOperatorInfo* pOperator = (SOperatorInfo*) param;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
SMultiwayMergeInfo* pAggInfo = pOperator->info;
SQueryRuntimeEnv *pRuntimeEnv = pOperator->pRuntimeEnv;
SOperatorInfo *upstream = pOperator->upstream;
while(1) {
SSDataBlock* pBlock = upstream->exec(upstream);
if (pBlock == NULL) {
break;
}
// not belongs to the same group, return the result of current group;
setInputDataBlock(pOperator, pAggInfo->binfo.pCtx, pBlock, TSDB_ORDER_ASC);
doExecuteFinalMergeRv(pAggInfo, pOperator->numOfOutput, pBlock, false);
}
pOperator->status = OP_EXEC_DONE;
setQueryStatus(pRuntimeEnv, QUERY_COMPLETED);
// finalizeQueryResult(pOperator, pAggInfo->binfo.pCtx, &pAggInfo->binfo.resultRowInfo, pAggInfo->binfo.rowCellInfoOffset);
pAggInfo->binfo.pRes->info.rows = getNumOfResult(pRuntimeEnv, pAggInfo->binfo.pCtx, pOperator->numOfOutput);
pOperator->status = OP_EXEC_DONE;
setQueryStatus(pRuntimeEnv, QUERY_COMPLETED);
return pAggInfo->binfo.pRes;
}

34
src/client/src/tscServer.c
View File

@ -1573,12 +1573,40 @@ int tscProcessRetrieveLocalMergeRsp(SSqlObj *pSql) {
return code;
}
pRes->code = tscDoLocalMerge(pSql);
SQueryInfo *pQueryInfo = tscGetActiveQueryInfo(pCmd);
if (pQueryInfo->pQInfo == NULL) {
STableGroupInfo tableGroupInfo = {.numOfTables = 1, .pGroupList = taosArrayInit(1, POINTER_BYTES),};
tableGroupInfo.map = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
STableKeyInfo tableKeyInfo = {.pTable = NULL, .lastKey = INT64_MIN};
SArray* group = taosArrayInit(1, sizeof(STableKeyInfo));
taosArrayPush(group, &tableKeyInfo);
taosArrayPush(tableGroupInfo.pGroupList, &group);
SExprInfo* list = calloc(tscSqlExprNumOfExprs(pQueryInfo), sizeof(SExprInfo));
for(int32_t i = 0; i < tscSqlExprNumOfExprs(pQueryInfo); ++i) {
SExprInfo* pExprInfo = tscSqlExprGet(pQueryInfo, i);
list[i] = *pExprInfo;
}
pQueryInfo->pQInfo = createQueryInfoFromQueryNode(pQueryInfo, list, &tableGroupInfo, NULL, NULL, pRes->pLocalMerger, MERGE_STAGE);
}
uint64_t localQueryId = 0;
SMultiwayMergeInfo* pInfo = (SMultiwayMergeInfo*) pQueryInfo->pQInfo->runtimeEnv.proot->info;
pInfo->pMerge = pRes->pLocalMerger;
qTableQuery(pQueryInfo->pQInfo, &localQueryId);
SSDataBlock* p = pQueryInfo->pQInfo->runtimeEnv.outputBuf;
pRes->numOfRows = (p != NULL)? p->info.rows: 0;
//pRes->code = tscDoLocalMerge(pSql);
if (pRes->code == TSDB_CODE_SUCCESS && pRes->numOfRows > 0) {
SQueryInfo *pQueryInfo = tscGetActiveQueryInfo(pCmd);
tscCreateResPointerInfo(pRes, pQueryInfo);
tscSetResRawPtr(pRes, pQueryInfo);
tscSetResRawPtrRv(pRes, pQueryInfo, p);
// tscSetResRawPtr(pRes, pQueryInfo);
}
pRes->row = 0;

22
src/client/src/tscSubquery.c
View File

@ -3467,7 +3467,8 @@ static UNUSED_FUNC bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) {
SQueryInfo *pQueryInfo1 = tscGetQueryInfo(&pSql->pSubs[i]->cmd, 0);
if ((pRes1->row >= pRes1->numOfRows && tscHasReachLimitation(pQueryInfo1, pRes1) &&
tscIsProjectionQuery(pQueryInfo1)) || (pRes1->numOfRows == 0)) {
tscIsProjectionQuery(pQueryInfo1)) ||
(pRes1->numOfRows == 0)) {
hasData = false;
break;
}
@ -3477,7 +3478,8 @@ static UNUSED_FUNC bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) {
return hasData;
}
void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr) {
void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo,
SOperatorInfo* pSourceOperator, char* sql, void* merger, int32_t stage) {
assert(pQueryInfo != NULL);
int16_t numOfOutput = pQueryInfo->fieldsInfo.numOfOutput;
@ -3493,7 +3495,7 @@ void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, ST
SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
pRuntimeEnv->pQueryAttr = pQueryAttr;
tscCreateQueryFromQueryInfo(pQueryInfo, pQueryAttr, addr);
tscCreateQueryFromQueryInfo(pQueryInfo, pQueryAttr, NULL);
pQueryAttr->tableGroupInfo = *pTableGroupInfo;
@ -3580,15 +3582,17 @@ void* createQueryInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, ST
tfree(pExprs);
SArray* pa = createExecOperatorPlan(pQueryAttr);
SArray* pa = NULL;
if (stage == MASTER_SCAN) {
pa = createExecOperatorPlan(pQueryAttr);
} else {
pa = createGlobalMergePlan(pQueryAttr);
}
STsBufInfo bufInfo = {0};
SQueryParam param = {.pOperator = pa};
/*int32_t code = */initQInfo(&bufInfo, NULL, pQInfo, &param, NULL, 0);
pQInfo->runtimeEnv.proot->upstream = pOperator;
qTableQuery(pQInfo, NULL);
/*int32_t code = */initQInfo(&bufInfo, NULL, pQInfo, &param, NULL, 0, merger);
// pQInfo->runtimeEnv.proot->upstream = pSourceOperator;
return pQInfo;
_cleanup:

131
src/client/src/tscUtil.c
View File

@ -522,6 +522,75 @@ void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
}
}
void tscSetResRawPtrRv(SSqlRes* pRes, SQueryInfo* pQueryInfo, SSDataBlock* pBlock) {
assert(pRes->numOfCols > 0);
int32_t offset = 0;
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SInternalField* pInfo = (SInternalField*)TARRAY_GET_ELEM(pQueryInfo->fieldsInfo.internalField, i);
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
pRes->urow[i] = pColData->pData + offset * pColData->info.bytes;
pRes->length[i] = pInfo->field.bytes;
offset += pInfo->field.bytes;
// generated the user-defined column result
if (pInfo->pExpr->pExpr == NULL && TSDB_COL_IS_UD_COL(pInfo->pExpr->base.colInfo.flag)) {
if (pInfo->pExpr->base.param[1].nType == TSDB_DATA_TYPE_NULL) {
setNullN(pRes->urow[i], pInfo->field.type, pInfo->field.bytes, (int32_t) pRes->numOfRows);
} else {
if (pInfo->field.type == TSDB_DATA_TYPE_NCHAR || pInfo->field.type == TSDB_DATA_TYPE_BINARY) {
assert(pInfo->pExpr->base.param[1].nLen <= pInfo->field.bytes);
for (int32_t k = 0; k < pRes->numOfRows; ++k) {
char* p = ((char**)pRes->urow)[i] + k * pInfo->field.bytes;
memcpy(varDataVal(p), pInfo->pExpr->base.param[1].pz, pInfo->pExpr->base.param[1].nLen);
varDataSetLen(p, pInfo->pExpr->base.param[1].nLen);
}
} else {
for (int32_t k = 0; k < pRes->numOfRows; ++k) {
char* p = ((char**)pRes->urow)[i] + k * pInfo->field.bytes;
memcpy(p, &pInfo->pExpr->base.param[1].i64, pInfo->field.bytes);
}
}
}
} else if (pInfo->field.type == TSDB_DATA_TYPE_NCHAR) {
// convert unicode to native code in a temporary buffer extra one byte for terminated symbol
pRes->buffer[i] = realloc(pRes->buffer[i], pInfo->field.bytes * pRes->numOfRows);
// string terminated char for binary data
memset(pRes->buffer[i], 0, pInfo->field.bytes * pRes->numOfRows);
char* p = pRes->urow[i];
for (int32_t k = 0; k < pRes->numOfRows; ++k) {
char* dst = pRes->buffer[i] + k * pInfo->field.bytes;
if (isNull(p, TSDB_DATA_TYPE_NCHAR)) {
memcpy(dst, p, varDataTLen(p));
} else if (varDataLen(p) > 0) {
int32_t length = taosUcs4ToMbs(varDataVal(p), varDataLen(p), varDataVal(dst));
varDataSetLen(dst, length);
if (length == 0) {
tscError("charset:%s to %s. val:%s convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, (char*)p);
}
} else {
varDataSetLen(dst, 0);
}
p += pInfo->field.bytes;
}
memcpy(pRes->urow[i], pRes->buffer[i], pInfo->field.bytes * pRes->numOfRows);
}
}
}
static SColumnInfo* extractColumnInfoFromResult(STableMeta* pTableMeta, SArray* pTableCols) {
int32_t numOfCols = taosArrayGetSize(pTableCols);
SColumnInfo* pColInfo = calloc(numOfCols, sizeof(SColumnInfo));
@ -626,8 +695,7 @@ void prepareInputDataFromUpstream(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
SOperatorInfo* pSourceOptr = createDummyInputOperator((char*)pRes, pSchema, numOfOutput);
SQInfo* pQInfo = createQueryInfoFromQueryNode(px, exprInfo, &tableGroupInfo, pSourceOptr, NULL, NULL);
//printf("%p\n", pQInfo);
SQInfo* pQInfo = createQueryInfoFromQueryNode(px, exprInfo, &tableGroupInfo, pSourceOptr, NULL, NULL, MASTER_SCAN);
SSDataBlock* pres = pQInfo->runtimeEnv.outputBuf;
// build result
@ -3250,6 +3318,60 @@ static int32_t createSecondaryExpr(SQueryAttr* pQueryAttr, SQueryInfo* pQueryInf
return TSDB_CODE_SUCCESS;
}
static int32_t createGlobalAggregateExpr(SQueryAttr* pQueryAttr, SQueryInfo* pQueryInfo) {
assert(tscIsTwoStageSTableQuery(pQueryInfo, 0));
pQueryAttr->numOfExpr3 = tscNumOfFields(pQueryInfo);
pQueryAttr->pExpr3 = calloc(pQueryAttr->numOfExpr3, sizeof(SExprInfo));
if (pQueryAttr->pExpr3 == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < pQueryAttr->numOfExpr3; ++i) {
SExprInfo* pExpr = &pQueryAttr->pExpr1[i];
SSqlExpr* pse = &pQueryAttr->pExpr3[i].base;
*pse = pExpr->base;
pse->colInfo.colId = pExpr->base.resColId;
pse->colInfo.colIndex = i;
pse->colType = pExpr->base.resType;
pse->colBytes = pExpr->base.resBytes;
for (int32_t j = 0; j < pExpr->base.numOfParams; ++j) {
tVariantAssign(&pse->param[j], &pExpr->base.param[j]);
}
}
{
for (int32_t i = 0; i < pQueryAttr->numOfExpr3; ++i) {
SExprInfo* pExpr = &pQueryAttr->pExpr1[i];
SSqlExpr* pse = &pQueryAttr->pExpr3[i].base;
// the final result size and type in the same as query on single table.
// so here, set the flag to be false;
int32_t inter = 0;
int32_t functionId = pExpr->base.functionId;
if (functionId >= TSDB_FUNC_TS && functionId <= TSDB_FUNC_DIFF) {
continue;
}
if (functionId == TSDB_FUNC_FIRST_DST) {
functionId = TSDB_FUNC_FIRST;
} else if (functionId == TSDB_FUNC_LAST_DST) {
functionId = TSDB_FUNC_LAST;
} else if (functionId == TSDB_FUNC_STDDEV_DST) {
functionId = TSDB_FUNC_STDDEV;
}
getResultDataInfo(pExpr->base.colType, pExpr->base.colBytes, functionId, 0, &pse->resType,
&pse->resBytes, &inter, 0, false);
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t createTagColumnInfo(SQueryAttr* pQueryAttr, SQueryInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo) {
if (pTableMetaInfo->tagColList == NULL) {
return TSDB_CODE_SUCCESS;
@ -3352,6 +3474,11 @@ int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAt
return code;
}
// global aggregate query
if (pQueryAttr->stableQuery && (pQueryAttr->simpleAgg || pQueryAttr->interval.interval > 0) && tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
createGlobalAggregateExpr(pQueryAttr, pQueryInfo);
}
// tag column info
code = createTagColumnInfo(pQueryAttr, pQueryInfo, pTableMetaInfo);
if (code != TSDB_CODE_SUCCESS) {

3
src/query/inc/qAggMain.h
View File

@ -210,9 +210,6 @@ typedef struct SAggFunctionInfo {
void (*xFunction)(SQLFunctionCtx *pCtx); // blocks version function
void (*xFunctionF)(SQLFunctionCtx *pCtx, int32_t position); // single-row function version, todo merge with blockwise function
// some sql function require scan data twice or more, e.g.,stddev, percentile
void (*xNextStep)(SQLFunctionCtx *pCtx);
// finalizer must be called after all xFunction has been executed to generated final result.
void (*xFinalize)(SQLFunctionCtx *pCtx);
void (*mergeFunc)(SQLFunctionCtx *pCtx);

33
src/query/inc/qExecutor.h
View File

@ -212,9 +212,13 @@ typedef struct SQueryAttr {
int32_t maxTableColumnWidth;
int32_t tagLen; // tag value length of current query
SSqlGroupbyExpr* pGroupbyExpr;
SExprInfo* pExpr1;
SExprInfo* pExpr2;
int32_t numOfExpr2;
SExprInfo* pExpr3;
int32_t numOfExpr3;
SColumnInfo* colList;
SColumnInfo* tagColList;
int32_t numOfFilterCols;
@ -290,7 +294,8 @@ enum OPERATOR_TYPE_E {
OP_MultiTableAggregate = 14,
OP_MultiTableTimeInterval = 15,
OP_DummyInput = 16, //TODO remove it after fully refactor.
OP_MultiwayMerge = 17, // multi-way merge process for partial results from different vnodes
OP_MultiwaySort = 17, // multi-way data merge into one input stream.
OP_GlobalAggregate = 18, // global merge for the multi-way data sources.
};
typedef struct SOperatorInfo {
@ -412,10 +417,6 @@ typedef struct SLimitOperatorInfo {
int64_t total;
} SLimitOperatorInfo;
typedef struct SOffsetOperatorInfo {
int64_t offset;
} SOffsetOperatorInfo;
typedef struct SFillOperatorInfo {
SFillInfo *pFillInfo;
SSDataBlock *pRes;
@ -436,6 +437,17 @@ typedef struct SSWindowOperatorInfo {
int32_t start; // start row index
} SSWindowOperatorInfo;
struct SLocalMerger;
typedef struct SMultiwayMergeInfo {
struct SLocalMerger *pMerge;
SOptrBasicInfo binfo;
int64_t seed;
char **prevRow;
bool hasPrev;
SArray *orderColumnList;
} SMultiwayMergeInfo;
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv, int32_t repeatTime, int32_t reverseTime);
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv, int32_t repeatTime);
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv);
@ -451,7 +463,14 @@ SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SO
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv);
SOperatorInfo* createMultiwaySortOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput,
int32_t numOfRows, void* merger);
SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, int32_t* orderColumn, int32_t numOfOrder);
SSDataBlock* doGlobalAggregate(void* param);
SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows);
void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
int32_t getNumOfResult(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx* pCtx, int32_t numOfOutput);
void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
void freeParam(SQueryParam *param);
int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param);
@ -466,7 +485,7 @@ SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr
SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, char* sql, uint64_t *qId);
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, SQInfo* pQInfo, SQueryParam* param, char* start,
int32_t prevResultLen);
int32_t prevResultLen, void* merger);
void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);

3
src/query/inc/qExtbuffer.h
View File

@ -237,6 +237,9 @@ int32_t compare_a(tOrderDescriptor *, int32_t numOfRow1, int32_t s1, char *data1
int32_t compare_d(tOrderDescriptor *, int32_t numOfRow1, int32_t s1, char *data1, int32_t numOfRow2, int32_t s2,
char *data2);
struct SSDataBlock;
int32_t compare_aRv(struct SSDataBlock* pBlock, int16_t* colIndex, int32_t numOfCols, int32_t rowIndex, char** buffer, int32_t order);
#ifdef __cplusplus
}
#endif

1
src/query/inc/qPlan.h
View File

@ -19,5 +19,6 @@
//TODO refactor
SArray* createTableScanPlan(SQueryAttr* pQueryAttr);
SArray* createExecOperatorPlan(SQueryAttr* pQueryAttr);
SArray* createGlobalMergePlan(SQueryAttr* pQueryAttr);
#endif // TDENGINE_QPLAN_H

47
src/query/src/qAggMain.c
View File

@ -375,12 +375,6 @@ int32_t isValidFunction(const char* name, int32_t len) {
return -1;
}
// set the query flag to denote that query is completed
static void no_next_step(SQLFunctionCtx *pCtx) {
SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
pResInfo->complete = true;
}
static bool function_setup(SQLFunctionCtx *pCtx) {
SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
if (pResInfo->initialized) {
@ -1540,7 +1534,7 @@ static void stddev_function_f(SQLFunctionCtx *pCtx, int32_t index) {
}
}
static void stddev_next_step(SQLFunctionCtx *pCtx) {
static UNUSED_FUNC void stddev_next_step(SQLFunctionCtx *pCtx) {
/*
* the stddevInfo and the average info struct share the same buffer area
* And the position of each element in their struct is exactly the same matched
@ -2907,7 +2901,7 @@ static void percentile_finalizer(SQLFunctionCtx *pCtx) {
doFinalizer(pCtx);
}
static void percentile_next_step(SQLFunctionCtx *pCtx) {
static UNUSED_FUNC void percentile_next_step(SQLFunctionCtx *pCtx) {
SResultRowCellInfo * pResInfo = GET_RES_INFO(pCtx);
SPercentileInfo *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
@ -4891,7 +4885,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
count_function,
count_function_f,
no_next_step,
doFinalizer,
count_func_merge,
countRequired,
@ -4905,7 +4898,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
sum_function,
sum_function_f,
no_next_step,
function_finalizer,
sum_func_merge,
statisRequired,
@ -4919,7 +4911,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
avg_function,
avg_function_f,
no_next_step,
avg_finalizer,
avg_func_merge,
statisRequired,
@ -4933,7 +4924,6 @@ SAggFunctionInfo aAggs[] = {{
min_func_setup,
min_function,
min_function_f,
no_next_step,
function_finalizer,
min_func_merge,
statisRequired,
@ -4947,7 +4937,6 @@ SAggFunctionInfo aAggs[] = {{
max_func_setup,
max_function,
max_function_f,
no_next_step,
function_finalizer,
max_func_merge,
statisRequired,
@ -4961,7 +4950,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
stddev_function,
stddev_function_f,
stddev_next_step,
stddev_finalizer,
noop1,
dataBlockRequired,
@ -4975,7 +4963,6 @@ SAggFunctionInfo aAggs[] = {{
percentile_function_setup,
percentile_function,
percentile_function_f,
percentile_next_step,
percentile_finalizer,
noop1,
dataBlockRequired,
@ -4989,7 +4976,6 @@ SAggFunctionInfo aAggs[] = {{
apercentile_function_setup,
apercentile_function,
apercentile_function_f,
no_next_step,
apercentile_finalizer,
apercentile_func_merge,
dataBlockRequired,
@ -5003,7 +4989,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
first_function,
first_function_f,
no_next_step,
function_finalizer,
noop1,
firstFuncRequired,
@ -5017,7 +5002,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
last_function,
last_function_f,
no_next_step,
function_finalizer,
noop1,
lastFuncRequired,
@ -5032,7 +5016,6 @@ SAggFunctionInfo aAggs[] = {{
first_last_function_setup,
last_row_function,
noop2,
no_next_step,
last_row_finalizer,
last_dist_func_merge,
dataBlockRequired,
@ -5047,7 +5030,6 @@ SAggFunctionInfo aAggs[] = {{
top_bottom_function_setup,
top_function,
top_function_f,
no_next_step,
top_bottom_func_finalizer,
top_func_merge,
dataBlockRequired,
@ -5062,7 +5044,6 @@ SAggFunctionInfo aAggs[] = {{
top_bottom_function_setup,
bottom_function,
bottom_function_f,
no_next_step,
top_bottom_func_finalizer,
bottom_func_merge,
dataBlockRequired,
@ -5076,7 +5057,6 @@ SAggFunctionInfo aAggs[] = {{
spread_function_setup,
spread_function,
spread_function_f,
no_next_step,
spread_function_finalizer,
spread_func_merge,
countRequired,
@ -5090,7 +5070,6 @@ SAggFunctionInfo aAggs[] = {{
twa_function_setup,
twa_function,
twa_function_f,
no_next_step,
twa_function_finalizer,
twa_function_copy,
dataBlockRequired,
@ -5104,7 +5083,6 @@ SAggFunctionInfo aAggs[] = {{
leastsquares_function_setup,
leastsquares_function,
leastsquares_function_f,
no_next_step,
leastsquares_finalizer,
noop1,
dataBlockRequired,
@ -5118,7 +5096,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
date_col_output_function,
date_col_output_function_f,
no_next_step,
doFinalizer,
copy_function,
noDataRequired,
@ -5132,7 +5109,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
noop1,
noop2,
no_next_step,
doFinalizer,
copy_function,
dataBlockRequired,
@ -5146,7 +5122,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
tag_function,
noop2,
no_next_step,
doFinalizer,
copy_function,
noDataRequired,
@ -5160,7 +5135,6 @@ SAggFunctionInfo aAggs[] = {{
ts_comp_function_setup,
ts_comp_function,
ts_comp_function_f,
no_next_step,
ts_comp_finalize,
copy_function,
dataBlockRequired,
@ -5174,7 +5148,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
tag_function,
tag_function_f,
no_next_step,
doFinalizer,
copy_function,
noDataRequired,
@ -5188,7 +5161,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
col_project_function,
col_project_function_f,
no_next_step,
doFinalizer,
copy_function,
dataBlockRequired,
@ -5202,7 +5174,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
tag_project_function,
tag_project_function_f,
no_next_step,
doFinalizer,
copy_function,
noDataRequired,
@ -5216,7 +5187,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
arithmetic_function,
arithmetic_function_f,
no_next_step,
doFinalizer,
copy_function,
dataBlockRequired,
@ -5230,7 +5200,6 @@ SAggFunctionInfo aAggs[] = {{
diff_function_setup,
diff_function,
diff_function_f,
no_next_step,
doFinalizer,
noop1,
dataBlockRequired,
@ -5245,7 +5214,6 @@ SAggFunctionInfo aAggs[] = {{
first_last_function_setup,
first_dist_function,
first_dist_function_f,
no_next_step,
function_finalizer,
first_dist_func_merge,
firstDistFuncRequired,
@ -5259,7 +5227,6 @@ SAggFunctionInfo aAggs[] = {{
first_last_function_setup,
last_dist_function,
last_dist_function_f,
no_next_step,
function_finalizer,
last_dist_func_merge,
lastDistFuncRequired,
@ -5273,7 +5240,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
stddev_dst_function,
stddev_dst_function_f,
no_next_step,
stddev_dst_finalizer,
stddev_dst_merge,
dataBlockRequired,
@ -5287,7 +5253,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
interp_function,
do_sum_f, // todo filter handle
no_next_step,
doFinalizer,
copy_function,
dataBlockRequired,
@ -5301,7 +5266,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
rate_function,
rate_function_f,
no_next_step,
rate_finalizer,
rate_func_copy,
dataBlockRequired,
@ -5315,7 +5279,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
irate_function,
irate_function_f,
no_next_step,
rate_finalizer,
rate_func_copy,
dataBlockRequired,
@ -5329,7 +5292,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
rate_function,
rate_function_f,
no_next_step,
sumrate_finalizer,
sumrate_func_merge,
dataBlockRequired,
@ -5343,7 +5305,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
irate_function,
irate_function_f,
no_next_step,
sumrate_finalizer,
sumrate_func_merge,
dataBlockRequired,
@ -5357,7 +5318,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
rate_function,
rate_function_f,
no_next_step,
sumrate_finalizer,
sumrate_func_merge,
dataBlockRequired,
@ -5371,7 +5331,6 @@ SAggFunctionInfo aAggs[] = {{
rate_function_setup,
irate_function,
irate_function_f,
no_next_step,
sumrate_finalizer,
sumrate_func_merge,
dataBlockRequired,
@ -5385,7 +5344,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
noop1,
noop2,
no_next_step,
noop1,
noop1,
dataBlockRequired,
@ -5399,7 +5357,6 @@ SAggFunctionInfo aAggs[] = {{
function_setup,
blockInfo_func,
noop2,
no_next_step,
blockinfo_func_finalizer,
block_func_merge,
dataBlockRequired,

165
src/query/src/qExecutor.c
View File

@ -180,7 +180,7 @@ static void doSetTableGroupOutputBuf(SQueryRuntimeEnv* pRuntimeEnv, SResultRowIn
int32_t groupIndex);
// setup the output buffer for each operator
static SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows) {
SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows) {
const static int32_t minSize = 8;
SSDataBlock *res = calloc(1, sizeof(SSDataBlock));
@ -351,7 +351,6 @@ static SResultRow *doPrepareResultRowFromKey(SQueryRuntimeEnv *pRuntimeEnv, SRes
prepareResultListBuffer(pResultRowInfo, pRuntimeEnv);
SResultRow *pResult = NULL;
if (p1 == NULL) {
pResult = getNewResultRow(pRuntimeEnv->pool);
int32_t ret = initResultRow(pResult);
@ -879,7 +878,7 @@ static void doSetInputDataBlockInfo(SOperatorInfo* pOperator, SQLFunctionCtx* pC
}
}
static void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order) {
void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order) {
if (pCtx[0].functionId == TSDB_FUNC_ARITHM) {
SArithmeticSupport* pSupport = (SArithmeticSupport*) pCtx[0].param[1].pz;
if (pSupport->colList == NULL) {
@ -1618,7 +1617,7 @@ static void* destroySQLFunctionCtx(SQLFunctionCtx* pCtx, int32_t numOfOutput) {
return NULL;
}
static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOfTables, SArray* pOperator) {
static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOfTables, SArray* pOperator, void* merger) {
qDebug("QInfo:%"PRIu64" setup runtime env", GET_QID(pRuntimeEnv));
SQueryAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
@ -1728,80 +1727,24 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
break;
}
case OP_MultiwaySort: {
pRuntimeEnv->proot = createMultiwaySortOperatorInfo(pRuntimeEnv, pQueryAttr->pExpr3, pQueryAttr->numOfExpr3,
4096, merger); // TODO hack it
break;
}
case OP_GlobalAggregate: {
pRuntimeEnv->proot = createGlobalAggregateOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr3,
pQueryAttr->numOfExpr3, &pQueryAttr->order.orderColId, 1);
break;
}
default: {
assert(0);
}
}
}
/*
if (onlyQueryTags(pQueryAttr)) { // do nothing for tags query
} else if (QUERY_IS_INTERVAL_QUERY(pQueryAttr)) {
if (pQueryAttr->stableQuery) {
pRuntimeEnv->proot = createMultiTableTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner,
pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
} else {
pRuntimeEnv->proot =
createTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
if (pQueryAttr->pExpr2 != NULL) {
pRuntimeEnv->proot =
createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr2, pQueryAttr->numOfExpr2);
}
if (pQueryAttr->fillType != TSDB_FILL_NONE && !pQueryAttr->pointInterpQuery) {
SOperatorInfo* pInfo = pRuntimeEnv->proot;
pRuntimeEnv->proot = createFillOperatorInfo(pRuntimeEnv, pInfo, pInfo->pExpr, pInfo->numOfOutput);
}
}
} else if (pQueryAttr->groupbyColumn) {
pRuntimeEnv->proot =
createGroupbyOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
if (pQueryAttr->pExpr2 != NULL) {
pRuntimeEnv->proot = createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr2, pQueryAttr->numOfExpr2);
}
} else if (pQueryAttr->sw.gap > 0) {
pRuntimeEnv->proot = createSWindowOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
if (pQueryAttr->pExpr2 != NULL) {
pRuntimeEnv->proot = createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr2, pQueryAttr->numOfExpr2);
}
} else if (pQueryAttr->simpleAgg) {
if (pQueryAttr->stableQuery && !pQueryAttr->tsCompQuery) {
pRuntimeEnv->proot =
createMultiTableAggOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
} else {
pRuntimeEnv->proot =
createAggregateOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
}
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
if (pQueryAttr->pExpr2 != NULL && !pQueryAttr->stableQuery) {
pRuntimeEnv->proot = createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr2, pQueryAttr->numOfExpr2);
}
} else { // diff/add/multiply/subtract/division
if (!onlyQueryTags(pQueryAttr)) {
pRuntimeEnv->proot =
createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->pTableScanner, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
}
}
if (pQueryAttr->limit.offset > 0) {
pRuntimeEnv->proot = createOffsetOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot);
}
if (pQueryAttr->limit.limit > 0) {
pRuntimeEnv->proot = createLimitOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot);
}
*/
return TSDB_CODE_SUCCESS;
_clean:
@ -3918,7 +3861,8 @@ static SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, in
return pFillCol;
}
int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, SArray* prevResult, void *tsdb, int32_t tbScanner, SArray* pOperator) {
int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, SArray* prevResult, void *tsdb, int32_t tbScanner, SArray* pOperator,
void* param) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQueryAttr *pQueryAttr = pQInfo->runtimeEnv.pQueryAttr;
@ -3981,7 +3925,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, SArray* prevResult, void *ts
// create runtime environment
int32_t numOfTables = (int32_t)pQueryAttr->tableGroupInfo.numOfTables;
pQInfo->summary.tableInfoSize += (numOfTables * sizeof(STableQueryInfo));
code = setupQueryRuntimeEnv(pRuntimeEnv, (int32_t) pQueryAttr->tableGroupInfo.numOfTables, pOperator);
code = setupQueryRuntimeEnv(pRuntimeEnv, (int32_t) pQueryAttr->tableGroupInfo.numOfTables, pOperator, param);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
@ -4355,6 +4299,55 @@ SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntime
return pOptr;
}
SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput,
int32_t* orderColumn, int32_t numOfOrder) {
SMultiwayMergeInfo* pInfo = calloc(1, sizeof(SMultiwayMergeInfo));
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
int32_t numOfRows =
(int32_t)(GET_ROW_PARAM_FOR_MULTIOUTPUT(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery));
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, numOfRows);
pInfo->binfo.pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// TODO refactor
int32_t len = 0;
for(int32_t i = 0; i < numOfOutput; ++i) {
len += pExpr[i].base.resBytes;
}
pInfo->prevRow = taosArrayInit(numOfOrder, (POINTER_BYTES * numOfOrder + len));
int32_t offset = POINTER_BYTES * numOfOutput;
for(int32_t i = 0; i < numOfOrder; ++i) {
pInfo->prevRow[i] = (char*)pInfo->prevRow + offset;
int32_t index = orderColumn[i];
offset += pExpr[index].base.resBytes;
}
pInfo->orderColumnList = taosArrayFromList(orderColumn, numOfOrder, sizeof(int32_t));
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
pInfo->seed = rand();
setDefaultOutputBuf(pRuntimeEnv, &pInfo->binfo, pInfo->seed);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
pOperator->name = "GlobalAggregate";
pOperator->operatorType = OP_GlobalAggregate;
pOperator->blockingOptr = true;
pOperator->status = OP_IN_EXECUTING;
pOperator->info = pInfo;
pOperator->upstream = upstream;
pOperator->pExpr = pExpr;
pOperator->numOfOutput = numOfOutput;
pOperator->pRuntimeEnv = pRuntimeEnv;
pOperator->exec = doGlobalAggregate;
pOperator->cleanup = destroyBasicOperatorInfo;
return pOperator;
}
static int32_t getTableScanOrder(STableScanInfo* pTableScanInfo) {
return pTableScanInfo->order;
}
@ -5122,6 +5115,25 @@ SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorIn
return pOperator;
}
SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream) {
SLimitOperatorInfo* pInfo = calloc(1, sizeof(SLimitOperatorInfo));
pInfo->limit = pRuntimeEnv->pQueryAttr->limit.limit;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
pOperator->name = "SLimitOperator";
pOperator->operatorType = OP_SLimit;
pOperator->blockingOptr = false;
pOperator->status = OP_IN_EXECUTING;
pOperator->upstream = upstream;
pOperator->exec = doLimit;
pOperator->info = pInfo;
pOperator->pRuntimeEnv = pRuntimeEnv;
return pOperator;
}
static SSDataBlock* doTagScan(void* param) {
SOperatorInfo* pOperator = (SOperatorInfo*) param;
if (pOperator->status == OP_EXEC_DONE) {
@ -6264,7 +6276,8 @@ bool isValidQInfo(void *param) {
return (sig == (uint64_t)pQInfo);
}
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, SQInfo* pQInfo, SQueryParam* param, char* start, int32_t prevResultLen) {
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, SQInfo* pQInfo, SQueryParam* param, char* start,
int32_t prevResultLen, void* merger) {
int32_t code = TSDB_CODE_SUCCESS;
SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
@ -6309,7 +6322,7 @@ int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, SQInfo* pQInfo, SQueryPara
}
// filter the qualified
if ((code = doInitQInfo(pQInfo, pTsBuf, prevResult, tsdb, param->tableScanOperator, param->pOperator)) != TSDB_CODE_SUCCESS) {
if ((code = doInitQInfo(pQInfo, pTsBuf, prevResult, tsdb, param->tableScanOperator, param->pOperator, merger)) != TSDB_CODE_SUCCESS) {
goto _error;
}

58
src/query/src/qExtbuffer.c
View File

@ -20,6 +20,7 @@
#include "taosdef.h"
#include "taosmsg.h"
#include "tulog.h"
#include "qExecutor.h"
#define COLMODEL_GET_VAL(data, schema, allrow, rowId, colId) \
(data + (schema)->pFields[colId].offset * (allrow) + (rowId) * (schema)->pFields[colId].field.bytes)
@ -351,6 +352,18 @@ static FORCE_INLINE int32_t primaryKeyComparator(int64_t f1, int64_t f2, int32_t
}
}
static int32_t tsCompareFunc(TSKEY k1, TSKEY k2, int32_t order) {
if (k1 == k2) {
return 0;
}
if (order == TSDB_ORDER_DESC) {
return (k1 < k2)? 1:-1;
} else {
return (k1 < k2)? -1:1;
}
}
static FORCE_INLINE int32_t columnValueAscendingComparator(char *f1, char *f2, int32_t type, int32_t bytes) {
switch (type) {
case TSDB_DATA_TYPE_INT: {
@ -451,6 +464,51 @@ int32_t compare_a(tOrderDescriptor *pDescriptor, int32_t numOfRows1, int32_t s1,
return 0;
}
int32_t compare_aRv(SSDataBlock* pBlock, int16_t* colIndex, int32_t numOfCols, int32_t rowIndex, char** buffer, int32_t order) {
for (int32_t i = 0; i < numOfCols; ++i) {
int32_t index = colIndex[i];
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, index);
char* data = pColInfo->pData + rowIndex * pColInfo->info.bytes;
if (pColInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
int32_t ret = tsCompareFunc(GET_INT64_VAL(data), GET_INT64_VAL(buffer[i]), order);
if (ret == 0) {
continue; // The timestamps are identical
} else {
return ret;
}
} else {
int32_t ret = columnValueAscendingComparator(data, buffer[i], pColInfo->info.type, pColInfo->info.bytes);
if (ret == 0) {
continue;
} else {
return ret;
}
}
// char *f1 = COLMODEL_GET_VAL(data1, pDescriptor->pColumnModel, numOfRows1, s1, colIdx);
// char *f2 = COLMODEL_GET_VAL(data2, pDescriptor->pColumnModel, numOfRows2, s2, colIdx);
// if (pDescriptor->pColumnModel->pFields[colIdx].field.type == TSDB_DATA_TYPE_TIMESTAMP) {
// int32_t ret = primaryKeyComparator(*(int64_t *)f1, *(int64_t *)f2, colIdx, pDescriptor->tsOrder);
// if (ret == 0) {
// continue;
// } else {
// return ret;
// }
// } else {
// SSchemaEx *pSchema = &pDescriptor->pColumnModel->pFields[colIdx];
// int32_t ret = columnValueAscendingComparator(f1, f2, pSchema->field.type, pSchema->field.bytes);
// if (ret == 0) {
// continue;
// } else {
// return ret;
// }
// }
}
return 0;
}
int32_t compare_d(tOrderDescriptor *pDescriptor, int32_t numOfRows1, int32_t s1, char *data1, int32_t numOfRows2,
int32_t s2, char *data2) {
assert(numOfRows1 == numOfRows2);

33
src/query/src/qPlan.c
View File

@ -125,4 +125,37 @@ SArray* createExecOperatorPlan(SQueryAttr* pQueryAttr) {
return plan;
}
SArray* createGlobalMergePlan(SQueryAttr* pQueryAttr) {
SArray* plan = taosArrayInit(4, sizeof(int32_t));
if (!pQueryAttr->stableQuery) {
return plan;
}
// todo: exchange operator?
int32_t op = OP_MultiwaySort;
taosArrayPush(plan, &op);
// fill operator
if (pQueryAttr->fillType != TSDB_FILL_NONE && (!pQueryAttr->pointInterpQuery)) {
op = OP_Fill;
taosArrayPush(plan, &op);
}
// arithmetic operator
if (!pQueryAttr->simpleAgg && pQueryAttr->interval.interval == 0) {
op = OP_Arithmetic;
taosArrayPush(plan, &op);
} else {
op = OP_GlobalAggregate;
taosArrayPush(plan, &op);
}
// limit/offset operator
if (pQueryAttr->limit.limit > 0 || pQueryAttr->limit.offset > 0) {
op = OP_SLimit;
taosArrayPush(plan, &op);
}
return plan;
}

2
src/query/src/queryMain.c
View File

@ -172,7 +172,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
goto _over;
}
code = initQInfo(&pQueryMsg->tsBuf, tsdb, *pQInfo, &param, (char*)pQueryMsg, pQueryMsg->prevResultLen);
code = initQInfo(&pQueryMsg->tsBuf, tsdb, *pQInfo, &param, (char*)pQueryMsg, pQueryMsg->prevResultLen, NULL);
_over:
if (param.pGroupbyExpr != NULL) {