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

[TD-1870]

This commit is contained in:
Haojun Liao 2020-11-02 14:10:15 +08:00
parent 2314b1952b
commit 5bbd7dd9d6
11 changed files with 379 additions and 399 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -66,7 +66,7 @@ typedef struct SLocalReducer {
SFillInfo* pFillInfo; // interpolation support structure
char * pFinalRes; // result data after interpo
tFilePage * discardData;
SResultInfo * pResInfo;
SResultRowCellInfo * pResInfo;
bool discard;
int32_t offset; // limit offset value
bool orderPrjOnSTable; // projection query on stable

384
src/client/src/tscFunctionImpl.c
View File

File diff suppressed because it is too large Load Diff

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

@ -99,12 +99,12 @@ static void tscInitSqlContext(SSqlCmd *pCmd, SLocalReducer *pReducer, tOrderDesc
pCtx->param[1].i64Key = pQueryInfo->order.orderColId;
}
SResultInfo *pResInfo = &pReducer->pResInfo[i];
pResInfo->bufLen = pExpr->interBytes;
pResInfo->interResultBuf = calloc(1, (size_t) pResInfo->bufLen);
// SResultRowCellInfo *pResInfo = &pReducer->pResInfo[i];
pCtx->interBufBytes = pExpr->interBytes;
// pResInfo->interResultBuf = calloc(1, (size_t) pCtx->interBufBytes);
pCtx->resultInfo = &pReducer->pResInfo[i];
pCtx->resultInfo->superTableQ = true;
pCtx->stableQuery = true;
}
int16_t n = 0;
@ -345,7 +345,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
size_t numOfCols = tscSqlExprNumOfExprs(pQueryInfo);
pReducer->pTempBuffer->num = 0;
pReducer->pResInfo = calloc(numOfCols, sizeof(SResultInfo));
pReducer->pResInfo = calloc(numOfCols, sizeof(SResultRowCellInfo));
tscCreateResPointerInfo(pRes, pQueryInfo);
tscInitSqlContext(pCmd, pReducer, pDesc);
@ -512,7 +512,7 @@ void tscDestroyLocalReducer(SSqlObj *pSql) {
if (pLocalReducer->pResInfo != NULL) {
size_t num = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < num; ++i) {
taosTFree(pLocalReducer->pResInfo[i].interResultBuf);
// taosTFree(pLocalReducer->pResInfo[i].interResultBuf);
}
taosTFree(pLocalReducer->pResInfo);
@ -1072,7 +1072,7 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx)
continue;
}
SResultInfo* pResInfo = GET_RES_INFO(&pCtx[j]);
SResultRowCellInfo* pResInfo = GET_RES_INFO(&pCtx[j]);
if (maxOutput < pResInfo->numOfRes) {
maxOutput = pResInfo->numOfRes;
}

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

@ -61,14 +61,14 @@ typedef struct SSqlGroupbyExpr {
int16_t orderType; // order by type: asc/desc
} SSqlGroupbyExpr;
typedef struct SWindowResult {
typedef struct SResultRow {
int32_t pageId; // pageId & rowId is the position of current result in disk-based output buffer
int32_t rowId:15;
bool closed:1; // this result status: closed or opened
uint16_t numOfRows; // number of rows of current time window
SResultInfo* resultInfo; // For each result column, there is a resultInfo
SResultRowCellInfo* pCellInfo; // For each result column, there is a resultInfo
union {STimeWindow win; char* key;}; // start key of current time window
} SWindowResult;
} SResultRow;
/**
* If the number of generated results is greater than this value,
@ -82,7 +82,7 @@ typedef struct SResultRec {
} SResultRec;
typedef struct SWindowResInfo {
SWindowResult** pResult; // result list
SResultRow** pResult; // result list
int16_t type; // data type for hash key
int32_t capacity; // max capacity
int32_t curIndex; // current start active index
@ -169,11 +169,11 @@ typedef struct SQuery {
typedef struct SQueryRuntimeEnv {
jmp_buf env;
SResultInfo* resultInfo; // todo refactor to merge with SWindowResInfo
SResultRowCellInfo* resultInfo; // todo refactor to merge with SWindowResInfo
SQuery* pQuery;
SQLFunctionCtx* pCtx;
int32_t numOfRowsPerPage;
int16_t offset[TSDB_MAX_COLUMNS];
uint16_t offset[TSDB_MAX_COLUMNS];
uint16_t scanFlag; // denotes reversed scan of data or not
SFillInfo* pFillInfo;
SWindowResInfo windowResInfo;
@ -192,6 +192,8 @@ typedef struct SQueryRuntimeEnv {
SHashObj* pWindowHashTable; // quick locate the window object for each result
char* keyBuf; // window key buffer
SWindowResultPool* pool; // window result object pool
int32_t* rowCellInfoOffset;// offset value for each row result cell info
} SQueryRuntimeEnv;
enum {

13
src/query/inc/qUtil.h
View File

@ -26,8 +26,9 @@
int32_t getOutputInterResultBufSize(SQuery* pQuery);
void clearTimeWindowResBuf(SQueryRuntimeEnv* pRuntimeEnv, SWindowResult* pOneOutputRes);
void copyTimeWindowResBuf(SQueryRuntimeEnv* pRuntimeEnv, SWindowResult* dst, const SWindowResult* src);
void clearResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pRow);
void copyResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* dst, const SResultRow* src);
SResultRowCellInfo* getResultCell(SQueryRuntimeEnv* pRuntimeEnv, const SResultRow* pRow, int32_t index);
int32_t initWindowResInfo(SWindowResInfo* pWindowResInfo, SQueryRuntimeEnv* pRuntimeEnv, int32_t size,
int32_t threshold, int16_t type);
@ -42,7 +43,7 @@ void closeTimeWindow(SWindowResInfo* pWindowResInfo, int32_t slot);
void closeAllTimeWindow(SWindowResInfo* pWindowResInfo);
void removeRedundantWindow(SWindowResInfo *pWindowResInfo, TSKEY lastKey, int32_t order);
static FORCE_INLINE SWindowResult *getWindowResult(SWindowResInfo *pWindowResInfo, int32_t slot) {
static FORCE_INLINE SResultRow *getWindowResult(SWindowResInfo *pWindowResInfo, int32_t slot) {
assert(pWindowResInfo != NULL && slot >= 0 && slot < pWindowResInfo->size);
return pWindowResInfo->pResult[slot];
}
@ -52,9 +53,9 @@ static FORCE_INLINE SWindowResult *getWindowResult(SWindowResInfo *pWindowResInf
bool isWindowResClosed(SWindowResInfo *pWindowResInfo, int32_t slot);
int32_t createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize);
int32_t createQueryResultInfo(SQuery *pQuery, SResultRow *pResultRow);
static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWindowResult *pResult,
static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SResultRow *pResult,
tFilePage* page) {
assert(pResult != NULL && pRuntimeEnv != NULL);
@ -74,7 +75,7 @@ __filter_func_t *getValueFilterFuncArray(int32_t type);
size_t getWindowResultSize(SQueryRuntimeEnv* pRuntimeEnv);
SWindowResultPool* initWindowResultPool(size_t size);
SWindowResult* getNewWindowResult(SWindowResultPool* p);
SResultRow* getNewWindowResult(SWindowResultPool* p);
int64_t getWindowResultPoolMemSize(SWindowResultPool* p);
void* destroyWindowResultPool(SWindowResultPool* p);
int32_t getNumOfAllocatedWindowResult(SWindowResultPool* p);

30
src/query/inc/tsqlfunction.h
View File

@ -145,15 +145,14 @@ typedef struct SInterpInfoDetail {
int8_t primaryCol;
} SInterpInfoDetail;
typedef struct SResultInfo {
typedef struct SResultRowCellInfo {
int8_t hasResult; // result generated, not NULL value
bool initialized; // output buffer has been initialized
bool complete; // query has completed
bool superTableQ; // is super table query
uint32_t bufLen; // buffer size
uint64_t numOfRes; // num of output result in current buffer
void* interResultBuf; // output result buffer
} SResultInfo;
bool initialized; // output buffer has been initialized
bool complete; // query has completed
uint16_t numOfRes; // num of output result in current buffer
} SResultRowCellInfo;
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowCellInfo)))
struct SQLFunctionCtx;
@ -175,9 +174,11 @@ typedef struct SQLFunctionCtx {
int16_t inputBytes;
int16_t outputType;
int16_t outputBytes; // size of results, determined by function and input column data type
bool hasNull; // null value exist in current block
int16_t outputBytes; // size of results, determined by function and input column data type
int32_t interBufBytes; // internal buffer size
bool hasNull; // null value exist in current block
bool requireNull; // require null in some function
bool stableQuery;
int16_t functionId; // function id
void * aInputElemBuf;
char * aOutputBuf; // final result output buffer, point to sdata->data
@ -189,7 +190,8 @@ typedef struct SQLFunctionCtx {
void * ptsOutputBuf; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
SQLPreAggVal preAggVals;
tVariant tag;
SResultInfo *resultInfo;
SResultRowCellInfo *resultInfo;
SExtTagsInfo tagInfo;
} SQLFunctionCtx;
@ -274,16 +276,16 @@ bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const cha
(_r)->initialized = false; \
} while (0)
void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf);
//void setResultInfoBuf(SResultRowCellInfo *pResInfo, char* buf);
static FORCE_INLINE void initResultInfo(SResultInfo *pResInfo) {
static FORCE_INLINE void initResultInfo(SResultRowCellInfo *pResInfo, uint32_t bufLen) {
pResInfo->initialized = true; // the this struct has been initialized flag
pResInfo->complete = false;
pResInfo->hasResult = false;
pResInfo->numOfRes = 0;
memset(pResInfo->interResultBuf, 0, (size_t)pResInfo->bufLen);
memset(GET_ROWCELL_INTERBUF(pResInfo), 0, (size_t)bufLen);
}
#ifdef __cplusplus

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

@ -178,9 +178,9 @@ static void getNextTimeWindow(SQuery* pQuery, STimeWindow* tw) {
// todo move to utility
static int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *group);
static void setWindowResOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pResult);
static void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pResult);
static void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pResultInfo);
static void setWindowResOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult);
static void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult);
static void resetMergeResultBuf(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx *pCtx, SResultRow *pRow);
static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId);
static void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY *tsCol, SDataBlockInfo* pBlockInfo,
@ -255,7 +255,7 @@ int64_t getNumOfResult(SQueryRuntimeEnv *pRuntimeEnv) {
continue;
}
SResultInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
SResultRowCellInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
if (pResInfo != NULL && maxOutput < pResInfo->numOfRes) {
maxOutput = pResInfo->numOfRes;
}
@ -272,7 +272,7 @@ void updateNumOfResult(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOfRes) {
SQuery *pQuery = pRuntimeEnv->pQuery;
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
SResultInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
SResultRowCellInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
int16_t functionId = pRuntimeEnv->pCtx[j].functionId;
if (functionId == TSDB_FUNC_TS || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAGPRJ ||
@ -447,7 +447,7 @@ static bool hasNullValue(SColIndex* pColIndex, SDataStatis *pStatis, SDataStatis
return true;
}
static SWindowResult *doSetTimeWindowFromKey(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowResInfo, char *pData,
static SResultRow *doSetTimeWindowFromKey(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowResInfo, char *pData,
int16_t bytes, bool masterscan, uint64_t uid) {
SQuery *pQuery = pRuntimeEnv->pQuery;
@ -470,31 +470,27 @@ static SWindowResult *doSetTimeWindowFromKey(SQueryRuntimeEnv *pRuntimeEnv, SWin
}
char *t = realloc(pWindowResInfo->pResult, (size_t)(newCapacity * POINTER_BYTES));
// pRuntimeEnv->summary.winInfoSize += (newCapacity - pWindowResInfo->capacity) * sizeof(SWindowResult);
// pRuntimeEnv->summary.winInfoSize += (newCapacity - pWindowResInfo->capacity) * sizeof(SResultRow);
// pRuntimeEnv->summary.numOfTimeWindows += (newCapacity - pWindowResInfo->capacity);
if (t == NULL) {
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pWindowResInfo->pResult = (SWindowResult **)t;
pWindowResInfo->pResult = (SResultRow **)t;
int32_t inc = (int32_t)newCapacity - pWindowResInfo->capacity;
memset(&pWindowResInfo->pResult[pWindowResInfo->capacity], 0, POINTER_BYTES * inc);
pWindowResInfo->capacity = (int32_t)newCapacity;
}
// pRuntimeEnv->summary.winInfoSize += (pQuery->numOfOutput * sizeof(SResultInfo) + pRuntimeEnv->interBufSize) * inc;
SWindowResult* pResult = getNewWindowResult(pRuntimeEnv->pool);
// pRuntimeEnv->summary.winInfoSize += (pQuery->numOfOutput * sizeof(SResultRowCellInfo) + pRuntimeEnv->interBufSize) * inc;
SResultRow* pResult = getNewWindowResult(pRuntimeEnv->pool);
pWindowResInfo->pResult[pWindowResInfo->size] = pResult;
// for (int32_t i = pWindowResInfo->capacity; i < newCapacity; ++i) {
int32_t ret = createQueryResultInfo(pQuery, pResult, pRuntimeEnv->stableQuery, pRuntimeEnv->interBufSize);
if (ret != TSDB_CODE_SUCCESS) {
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
// }
// }
int32_t ret = createQueryResultInfo(pQuery, pResult);
if (ret != TSDB_CODE_SUCCESS) {
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
// add a new result set for a new group
pWindowResInfo->curIndex = pWindowResInfo->size++;
@ -522,7 +518,7 @@ static STimeWindow getActiveTimeWindow(SWindowResInfo *pWindowResInfo, int64_t t
}
} else {
int32_t slot = curTimeWindowIndex(pWindowResInfo);
SWindowResult* pWindowRes = getWindowResult(pWindowResInfo, slot);
SResultRow* pWindowRes = getWindowResult(pWindowResInfo, slot);
w = pWindowRes->win;
}
@ -558,7 +554,7 @@ static STimeWindow getActiveTimeWindow(SWindowResInfo *pWindowResInfo, int64_t t
return w;
}
static int32_t addNewWindowResultBuf(SWindowResult *pWindowRes, SDiskbasedResultBuf *pResultBuf, int32_t tid,
static int32_t addNewWindowResultBuf(SResultRow *pWindowRes, SDiskbasedResultBuf *pResultBuf, int32_t tid,
int32_t numOfRowsPerPage) {
if (pWindowRes->pageId != -1) {
return 0;
@ -608,7 +604,7 @@ static int32_t setWindowOutputBufByKey(SQueryRuntimeEnv *pRuntimeEnv, SWindowRes
assert(win->skey <= win->ekey);
SDiskbasedResultBuf *pResultBuf = pRuntimeEnv->pResultBuf;
SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&win->skey, TSDB_KEYSIZE, masterscan, pBockInfo->uid);
SResultRow *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&win->skey, TSDB_KEYSIZE, masterscan, pBockInfo->uid);
if (pWindowRes == NULL) {
*newWind = false;
@ -693,7 +689,7 @@ static int32_t doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKe
int64_t skey = TSKEY_INITIAL_VAL;
for (i = 0; i < pWindowResInfo->size; ++i) {
SWindowResult *pResult = pWindowResInfo->pResult[i];
SResultRow *pResult = pWindowResInfo->pResult[i];
if (pResult->closed) {
numOfClosed += 1;
continue;
@ -1118,7 +1114,7 @@ static int32_t setGroupResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, char *pDat
}
uint64_t uid = 0; // uid is always set to be 0.
SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, &pRuntimeEnv->windowResInfo, d, len, true, uid);
SResultRow *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, &pRuntimeEnv->windowResInfo, d, len, true, uid);
if (pWindowRes == NULL) {
return -1;
}
@ -1232,7 +1228,7 @@ static int32_t doTSJoinFilter(SQueryRuntimeEnv *pRuntimeEnv, int32_t offset) {
}
static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
SQuery* pQuery = pRuntimeEnv->pQuery;
// in case of timestamp column, always generated results.
@ -1519,7 +1515,8 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY
* top/bottom values emerge, so does diff function
*/
if (functionId == TSDB_FUNC_TWA) {
STwaInfo *pTWAInfo = GET_RES_INFO(pCtx)->interResultBuf;
SResultRowCellInfo* pInfo = GET_RES_INFO(pCtx);
STwaInfo *pTWAInfo = (STwaInfo*) GET_ROWCELL_INTERBUF(pInfo);
pTWAInfo->SKey = pQuery->window.skey;
pTWAInfo->EKey = pQuery->window.ekey;
}
@ -1533,7 +1530,9 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY
pCtx->preAggVals.statis.max = pBlockInfo->window.ekey;
}
} else if (functionId == TSDB_FUNC_INTERP) {
SInterpInfoDetail *pInterpInfo = GET_RES_INFO(pCtx)->interResultBuf;
SResultRowCellInfo* pInfo = GET_RES_INFO(pCtx);
SInterpInfoDetail *pInterpInfo = (SInterpInfoDetail *)GET_ROWCELL_INTERBUF(pInfo);
pInterpInfo->type = (int8_t)pQuery->fillType;
pInterpInfo->ts = pQuery->window.skey;
pInterpInfo->primaryCol = (colId == PRIMARYKEY_TIMESTAMP_COL_INDEX);
@ -1610,26 +1609,29 @@ static int32_t setCtxTagColumnInfo(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx
return TSDB_CODE_SUCCESS;
}
static FORCE_INLINE void setWindowResultInfo(SResultInfo *pResultInfo, SQuery *pQuery, bool isStableQuery, char* buf) {
char* p = buf;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
int32_t size = pQuery->pSelectExpr[i].interBytes;
setResultInfoBuf(&pResultInfo[i], size, isStableQuery, p);
p += size;
}
static FORCE_INLINE void setResultRowCellInfo(SQueryRuntimeEnv* pRuntimeEnv, SResultRow *pRow, char* buf) {
// SQuery* pQuery = pRuntimeEnv->pQuery;
//
// char* p = buf;
// for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
// int32_t size = pQuery->pSelectExpr[i].interBytes;
// SResultRowCellInfo* pInfo = getResultCell(pRuntimeEnv, pRow, i);
// setResultInfoBuf(pInfo, p);
// p += size;
// }
}
static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order) {
qDebug("QInfo:%p setup runtime env", GET_QINFO_ADDR(pRuntimeEnv));
SQuery *pQuery = pRuntimeEnv->pQuery;
size_t size = pRuntimeEnv->interBufSize + pQuery->numOfOutput * sizeof(SResultInfo);
size_t size = pRuntimeEnv->interBufSize + pQuery->numOfOutput * sizeof(SResultRowCellInfo);
pRuntimeEnv->resultInfo = calloc(1, size);
pRuntimeEnv->pCtx = (SQLFunctionCtx *)calloc(pQuery->numOfOutput, sizeof(SQLFunctionCtx));
pRuntimeEnv->rowCellInfoOffset = calloc(pQuery->numOfOutput, sizeof(int32_t));
if (pRuntimeEnv->resultInfo == NULL || pRuntimeEnv->pCtx == NULL) {
if (pRuntimeEnv->resultInfo == NULL || pRuntimeEnv->pCtx == NULL || pRuntimeEnv->rowCellInfoOffset == NULL) {
goto _clean;
}
@ -1675,6 +1677,8 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
pCtx->order = pQuery->order.order;
pCtx->functionId = pSqlFuncMsg->functionId;
pCtx->stableQuery = pRuntimeEnv->stableQuery;
pCtx->interBufBytes = pQuery->pSelectExpr[i].interBytes;
pCtx->numOfParams = pSqlFuncMsg->numOfParams;
for (int32_t j = 0; j < pCtx->numOfParams; ++j) {
@ -1704,13 +1708,15 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
if (i > 0) {
pRuntimeEnv->offset[i] = pRuntimeEnv->offset[i - 1] + pRuntimeEnv->pCtx[i - 1].outputBytes;
pRuntimeEnv->rowCellInfoOffset[i] = pRuntimeEnv->rowCellInfoOffset[i - 1] + sizeof(SResultRowCellInfo) + pQuery->pSelectExpr[i - 1].interBytes;
}
}
char* buf = (char*) pRuntimeEnv->resultInfo + sizeof(SResultInfo) * pQuery->numOfOutput;
// char* buf = (char*) pRuntimeEnv->resultInfo + sizeof(SResultRowCellInfo) * pQuery->numOfOutput;
// set the intermediate result output buffer
setWindowResultInfo(pRuntimeEnv->resultInfo, pQuery, pRuntimeEnv->stableQuery, buf);
// setResultRowCellInfo(pRuntimeEnv, pRuntimeEnv->resultInfo, NULL);
// if it is group by normal column, do not set output buffer, the output buffer is pResult
if (!pRuntimeEnv->groupbyNormalCol && !pRuntimeEnv->stableQuery) {
@ -2655,7 +2661,7 @@ void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, void *pTable, void *tsdb) {
}
}
static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SWindowResult *pWindowRes, bool mergeFlag) {
static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SResultRow *pWindowRes, bool mergeFlag) {
SQuery * pQuery = pRuntimeEnv->pQuery;
SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx;
@ -2829,14 +2835,14 @@ int32_t tableResultComparFn(const void *pLeft, const void *pRight, void *param)
}
SWindowResInfo *pWindowResInfo1 = &supporter->pTableQueryInfo[left]->windowResInfo;
SWindowResult * pWindowRes1 = getWindowResult(pWindowResInfo1, leftPos);
SResultRow * pWindowRes1 = getWindowResult(pWindowResInfo1, leftPos);
tFilePage *page1 = getResBufPage(pRuntimeEnv->pResultBuf, pWindowRes1->pageId);
char *b1 = getPosInResultPage(pRuntimeEnv, PRIMARYKEY_TIMESTAMP_COL_INDEX, pWindowRes1, page1);
TSKEY leftTimestamp = GET_INT64_VAL(b1);
SWindowResInfo *pWindowResInfo2 = &supporter->pTableQueryInfo[right]->windowResInfo;
SWindowResult * pWindowRes2 = getWindowResult(pWindowResInfo2, rightPos);
SResultRow * pWindowRes2 = getWindowResult(pWindowResInfo2, rightPos);
tFilePage *page2 = getResBufPage(pRuntimeEnv->pResultBuf, pWindowRes2->pageId);
char *b2 = getPosInResultPage(pRuntimeEnv, PRIMARYKEY_TIMESTAMP_COL_INDEX, pWindowRes2, page2);
@ -2957,7 +2963,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
pQuery->rec.rows += offset;
}
int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) {
int64_t getNumOfResultWindowRes(SQuery *pQuery, SResultRow *pResultRow) {
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
int32_t functionId = pQuery->pSelectExpr[j].base.functionId;
@ -2969,7 +2975,7 @@ int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) {
continue;
}
SResultInfo *pResultInfo = &pWindowRes->resultInfo[j];
SResultRowCellInfo *pResultInfo = &pResultRow->pCellInfo[j];
assert(pResultInfo != NULL);
if (pResultInfo->numOfRes > 0) {
@ -3038,18 +3044,19 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
SLoserTreeInfo *pTree = NULL;
tLoserTreeCreate(&pTree, numOfTables, &cs, tableResultComparFn);
SResultInfo *pResultInfo = calloc(pQuery->numOfOutput, sizeof(SResultInfo));
if (pResultInfo == NULL) {
SResultRow* pRow = calloc(1, getWindowResultSize(pRuntimeEnv));
if (pRow == NULL) {
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
char* buf = calloc(1, pRuntimeEnv->interBufSize);
if (buf == NULL) {
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
pRow->pCellInfo = (SResultRowCellInfo*) ((char*) pRow + sizeof(SResultRow));
// char* buf = (char*) pRow + sizeof(SResultRowCellInfo)*;
// if (buf == NULL) {
// longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
// }
setWindowResultInfo(pResultInfo, pQuery, pRuntimeEnv->stableQuery, buf);
resetMergeResultBuf(pQuery, pRuntimeEnv->pCtx, pResultInfo);
setResultRowCellInfo(pRuntimeEnv, pRow, NULL);
resetMergeResultBuf(pRuntimeEnv, pRuntimeEnv->pCtx, pRow);
pQInfo->groupResInfo.groupId = getGroupResultId(pQInfo->groupIndex);
@ -3064,8 +3071,8 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
taosTFree(pTableList);
taosTFree(posList);
taosTFree(pTree);
taosTFree(pResultInfo);
taosTFree(buf);
// taosTFree(pResultInfo);
// taosTFree(buf);
longjmp(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
@ -3073,7 +3080,7 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
int32_t pos = pTree->pNode[0].index;
SWindowResInfo *pWindowResInfo = &pTableList[pos]->windowResInfo;
SWindowResult *pWindowRes = getWindowResult(pWindowResInfo, cs.position[pos]);
SResultRow *pWindowRes = getWindowResult(pWindowResInfo, cs.position[pos]);
tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pWindowRes->pageId);
char *b = getPosInResultPage(pRuntimeEnv, PRIMARYKEY_TIMESTAMP_COL_INDEX, pWindowRes, page);
@ -3101,7 +3108,7 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
return -1;
}
resetMergeResultBuf(pQuery, pRuntimeEnv->pCtx, pResultInfo);
resetMergeResultBuf(pRuntimeEnv, pRuntimeEnv->pCtx, pRow);
}
doMerge(pRuntimeEnv, ts, pWindowRes, false);
@ -3123,7 +3130,7 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
}
} else {
// current page is not needed anymore
SWindowResult *pNextWindowRes = getWindowResult(pWindowResInfo, cs.position[pos]);
SResultRow *pNextWindowRes = getWindowResult(pWindowResInfo, cs.position[pos]);
if (pNextWindowRes->pageId != currentPageId) {
releaseResBufPage(pRuntimeEnv->pResultBuf, page);
}
@ -3140,7 +3147,7 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
taosTFree(pTree);
taosTFree(pTableList);
taosTFree(posList);
taosTFree(pResultInfo);
// taosTFree(pResultInfo);
return -1;
}
@ -3158,8 +3165,8 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
taosTFree(posList);
taosTFree(pTree);
taosTFree(pResultInfo);
taosTFree(buf);
// taosTFree(pResultInfo);
// taosTFree(buf);
return pQInfo->groupResInfo.numOfDataPages;
}
@ -3202,12 +3209,14 @@ int32_t flushFromResultBuf(SQueryRuntimeEnv* pRuntimeEnv, SGroupResInfo* pGroupR
return TSDB_CODE_SUCCESS;
}
void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pResultInfo) {
void resetMergeResultBuf(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx *pCtx, SResultRow *pRow) {
SQuery* pQuery = pRuntimeEnv->pQuery;
for (int32_t k = 0; k < pQuery->numOfOutput; ++k) {
pCtx[k].aOutputBuf = pQuery->sdata[k]->data - pCtx[k].outputBytes;
pCtx[k].size = 1;
pCtx[k].startOffset = 0;
pCtx[k].resultInfo = &pResultInfo[k];
pCtx[k].resultInfo = getResultCell(pRuntimeEnv, pRow, k);
pQuery->sdata[k]->num = 0;
}
@ -3253,7 +3262,7 @@ static void disableFuncInReverseScanImpl(SQInfo* pQInfo, SWindowResInfo *pWindow
continue;
}
SWindowResult *buf = getWindowResult(pWindowResInfo, i);
SResultRow *buf = getWindowResult(pWindowResInfo, i);
// open/close the specified query for each group result
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
@ -3261,9 +3270,9 @@ static void disableFuncInReverseScanImpl(SQInfo* pQInfo, SWindowResInfo *pWindow
if (((functId == TSDB_FUNC_FIRST || functId == TSDB_FUNC_FIRST_DST) && order == TSDB_ORDER_ASC) ||
((functId == TSDB_FUNC_LAST || functId == TSDB_FUNC_LAST_DST) && order == TSDB_ORDER_DESC)) {
buf->resultInfo[j].complete = false;
buf->pCellInfo[j].complete = false;
} else if (functId != TSDB_FUNC_TS && functId != TSDB_FUNC_TAG) {
buf->resultInfo[j].complete = true;
buf->pCellInfo[j].complete = true;
}
}
}
@ -3327,24 +3336,17 @@ void switchCtxOrder(SQueryRuntimeEnv *pRuntimeEnv) {
}
}
int32_t createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize) {
int32_t numOfCols = pQuery->numOfOutput;
// size_t size = numOfCols * sizeof(SResultInfo) + interBufSize;
pResultRow->resultInfo = (SResultInfo*)((char*)pResultRow + sizeof(SWindowResult));
// pResultRow->resultInfo = calloc(1, size);
// if (pResultRow->resultInfo == NULL) {
// return TSDB_CODE_QRY_OUT_OF_MEMORY;
// }
int32_t createQueryResultInfo(SQuery *pQuery, SResultRow *pResultRow) {
// int32_t numOfCols = pQuery->numOfOutput;
pResultRow->pCellInfo = (SResultRowCellInfo*)((char*)pResultRow + sizeof(SResultRow));
pResultRow->pageId = -1;
pResultRow->rowId = -1;
char* buf = (char*) pResultRow->resultInfo + numOfCols * sizeof(SResultInfo);
// char* buf = (char*) pResultRow->pCellInfo + numOfCols * sizeof(SResultRowCellInfo);
// set the intermediate result output buffer
setWindowResultInfo(pResultRow->resultInfo, pQuery, isSTableQuery, buf);
// setResultRowCellInfo(pRunimeEnv, pResultRow, buf);
return TSDB_CODE_SUCCESS;
}
@ -3409,7 +3411,7 @@ void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
int32_t functionId = pQuery->pSelectExpr[j].base.functionId;
pRuntimeEnv->pCtx[j].currentStage = 0;
SResultInfo* pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
SResultRowCellInfo* pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
if (pResInfo->initialized) {
continue;
}
@ -3478,7 +3480,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo;
for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
SWindowResult *pResult = getWindowResult(pWindowResInfo, i);
SResultRow *pResult = getWindowResult(pWindowResInfo, i);
if (!pResult->closed) {
continue;
}
@ -3492,7 +3494,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
}
aAggs[functId].xNextStep(&pRuntimeEnv->pCtx[j]);
SResultInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
SResultRowCellInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
toContinue |= (!pResInfo->complete);
}
@ -3505,7 +3507,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
}
aAggs[functId].xNextStep(&pRuntimeEnv->pCtx[j]);
SResultInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
SResultRowCellInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[j]);
toContinue |= (!pResInfo->complete);
}
@ -3704,7 +3706,7 @@ void finalizeQueryResult(SQueryRuntimeEnv *pRuntimeEnv) {
}
for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
SWindowResult *buf = pWindowResInfo->pResult[i];
SResultRow *buf = pWindowResInfo->pResult[i];
if (!isWindowResClosed(pWindowResInfo, i)) {
continue;
}
@ -3797,7 +3799,7 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) {
}
uint64_t uid = 0; // uid is always set to be 0
SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&groupIndex,
SResultRow *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&groupIndex,
sizeof(groupIndex), true, uid);
if (pWindowRes == NULL) {
return;
@ -3820,7 +3822,7 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) {
initCtxOutputBuf(pRuntimeEnv);
}
void setWindowResOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pResult) {
void setWindowResOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult) {
SQuery *pQuery = pRuntimeEnv->pQuery;
// Note: pResult->pos[i]->num == 0, there is only fixed number of results for each group
@ -3839,15 +3841,11 @@ void setWindowResOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pResult
* set the output buffer information and intermediate buffer
* not all queries require the interResultBuf, such as COUNT
*/
pCtx->resultInfo = &pResult->resultInfo[i];
// set super table query flag
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
pResInfo->superTableQ = pRuntimeEnv->stableQuery;
pCtx->resultInfo = &pResult->pCellInfo[i];
}
}
void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pResult) {
void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult) {
SQuery *pQuery = pRuntimeEnv->pQuery;
// Note: pResult->pos[i]->num == 0, there is only fixed number of results for each group
@ -3856,7 +3854,7 @@ void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
SQLFunctionCtx *pCtx = &pRuntimeEnv->pCtx[i];
pCtx->resultInfo = &pResult->resultInfo[i];
pCtx->resultInfo = &pResult->pCellInfo[i];
if (pCtx->resultInfo->initialized && pCtx->resultInfo->complete) {
continue;
}
@ -3869,12 +3867,6 @@ void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *
pCtx->ptsOutputBuf = pRuntimeEnv->pCtx[0].aOutputBuf;
}
/*
* set the output buffer information and intermediate buffer
* not all queries require the interResultBuf, such as COUNT
*/
pCtx->resultInfo->superTableQ = pRuntimeEnv->stableQuery; // set super table query flag
if (!pCtx->resultInfo->initialized) {
aAggs[functionId].init(pCtx);
}
@ -4015,7 +4007,7 @@ static int32_t doCopyToSData(SQInfo *pQInfo, SWindowResInfo *pResultInfo, int32_
qDebug("QInfo:%p start to copy data from windowResInfo to query buf", pQInfo);
int32_t totalSet = numOfClosedTimeWindow(pResultInfo);
SWindowResult** result = pResultInfo->pResult;
SResultRow** result = pResultInfo->pResult;
if (orderType == TSDB_ORDER_ASC) {
startIdx = pQInfo->groupIndex;
@ -4102,7 +4094,7 @@ static void updateWindowResNumOfRes(SQueryRuntimeEnv *pRuntimeEnv) {
}
for (int32_t i = 0; i < pRuntimeEnv->windowResInfo.size; ++i) {
SWindowResult *pResult = pRuntimeEnv->windowResInfo.pResult[i];
SResultRow *pResult = pRuntimeEnv->windowResInfo.pResult[i];
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
int32_t functionId = pRuntimeEnv->pCtx[j].functionId;
@ -4110,7 +4102,7 @@ static void updateWindowResNumOfRes(SQueryRuntimeEnv *pRuntimeEnv) {
continue;
}
pResult->numOfRows = (uint16_t)(MAX(pResult->numOfRows, pResult->resultInfo[j].numOfRes));
pResult->numOfRows = (uint16_t)(MAX(pResult->numOfRows, pResult->pCellInfo[j].numOfRes));
}
}
}
@ -4678,7 +4670,7 @@ static void enableExecutionForNextTable(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
SResultInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[i]);
SResultRowCellInfo *pResInfo = GET_RES_INFO(&pRuntimeEnv->pCtx[i]);
if (pResInfo != NULL) {
pResInfo->complete = false;
}
@ -5033,9 +5025,9 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
pWindowResInfo->pResult[i]->closed = true; // enable return all results for group by normal columns
SWindowResult *pResult = pWindowResInfo->pResult[i];
SResultRow *pResult = pWindowResInfo->pResult[i];
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
pResult->numOfRows = (uint16_t)(MAX(pResult->numOfRows, pResult->resultInfo[j].numOfRes));
pResult->numOfRows = (uint16_t)(MAX(pResult->numOfRows, pResult->pCellInfo[j].numOfRes));
}
}
@ -6297,7 +6289,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
calResultBufSize(pQuery);
for (int32_t col = 0; col < pQuery->numOfOutput; ++col) {
assert(pExprs[col].interBytes >= pExprs[col].bytes);
assert(pExprs[col].interBytes >= pExprs[col].bytes || pExprs[col].interBytes == 0);
// allocate additional memory for interResults that are usually larger then final results
size_t size = (size_t)((pQuery->rec.capacity + 1) * pExprs[col].bytes + pExprs[col].interBytes + sizeof(tFilePage));

6
src/query/src/qResultbuf.c
View File

@ -407,9 +407,9 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
}
if (pResultBuf->file != NULL) {
qDebug("QInfo:%p res output buffer closed, total:%.2f Kb, inmem size:%dbytes, file size:%"PRId64" bytes",
pResultBuf->handle, pResultBuf->totalBufSize/1024.0, listNEles(pResultBuf->lruList) * pResultBuf->pageSize,
pResultBuf->fileSize);
qDebug("QInfo:%p res output buffer closed, total:%.2f Kb, inmem size:%.2f Kb, file size:%.2f",
pResultBuf->handle, pResultBuf->totalBufSize/1024.0, listNEles(pResultBuf->lruList) * pResultBuf->pageSize / 1024.0,
pResultBuf->fileSize/1024.0);
fclose(pResultBuf->file);
} else {

59
src/query/src/qUtil.c
View File

@ -27,7 +27,7 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
size += pQuery->pSelectExpr[i].interBytes;
}
assert(size > 0);
assert(size >= 0);
return size;
}
@ -41,26 +41,12 @@ int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRun
pWindowResInfo->size = 0;
pWindowResInfo->prevSKey = TSKEY_INITIAL_VAL;
// SQueryCostInfo* pSummary = &pRuntimeEnv->summary;
pWindowResInfo->pResult = calloc(pWindowResInfo->capacity, POINTER_BYTES);
if (pWindowResInfo->pResult == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
pWindowResInfo->interval = pRuntimeEnv->pQuery->interval.interval;
// pSummary->winInfoSize += POINTER_BYTES * pWindowResInfo->capacity;
// pSummary->winInfoSize += (pRuntimeEnv->pQuery->numOfOutput * sizeof(SResultInfo) + pRuntimeEnv->interBufSize) * pWindowResInfo->capacity;
// pSummary->numOfTimeWindows = pWindowResInfo->capacity;
// for (int32_t i = 0; i < pWindowResInfo->capacity; ++i) {
// int32_t code = createQueryResultInfo(pRuntimeEnv->pQuery, pWindowResInfo->pResult[i], pRuntimeEnv->stableQuery, pRuntimeEnv->interBufSize);
// if (code != TSDB_CODE_SUCCESS) {
// return code;
// }
// }
return TSDB_CODE_SUCCESS;
}
@ -82,8 +68,8 @@ void resetTimeWindowInfo(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowR
}
for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
SWindowResult *pWindowRes = pWindowResInfo->pResult[i];
clearTimeWindowResBuf(pRuntimeEnv, pWindowRes);
SResultRow *pWindowRes = pWindowResInfo->pResult[i];
clearResultRow(pRuntimeEnv, pWindowRes);
}
pWindowResInfo->curIndex = -1;
@ -108,7 +94,7 @@ void clearFirstNTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
int16_t bytes = -1;
for (int32_t i = 0; i < num; ++i) {
SWindowResult *pResult = pWindowResInfo->pResult[i];
SResultRow *pResult = pWindowResInfo->pResult[i];
if (pResult->closed) { // remove the window slot from hash table
// todo refactor
@ -131,19 +117,19 @@ void clearFirstNTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
// clear all the closed windows from the window list
for (int32_t k = 0; k < remain; ++k) {
copyTimeWindowResBuf(pRuntimeEnv, pWindowResInfo->pResult[k], pWindowResInfo->pResult[num + k]);
copyResultRow(pRuntimeEnv, pWindowResInfo->pResult[k], pWindowResInfo->pResult[num + k]);
}
// move the unclosed window in the front of the window list
for (int32_t k = remain; k < pWindowResInfo->size; ++k) {
SWindowResult *pWindowRes = pWindowResInfo->pResult[k];
clearTimeWindowResBuf(pRuntimeEnv, pWindowRes);
SResultRow *pWindowRes = pWindowResInfo->pResult[k];
clearResultRow(pRuntimeEnv, pWindowRes);
}
pWindowResInfo->size = remain;
for (int32_t k = 0; k < pWindowResInfo->size; ++k) {
SWindowResult *pResult = pWindowResInfo->pResult[k];
SResultRow *pResult = pWindowResInfo->pResult[k];
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
key = varDataVal(pResult->key);
@ -240,7 +226,7 @@ void closeTimeWindow(SWindowResInfo *pWindowResInfo, int32_t slot) {
getWindowResult(pWindowResInfo, slot)->closed = true;
}
void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindowRes) {
void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pWindowRes) {
if (pWindowRes == NULL) {
return;
}
@ -248,7 +234,7 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pWindowRes->pageId);
for (int32_t i = 0; i < pRuntimeEnv->pQuery->numOfOutput; ++i) {
SResultInfo *pResultInfo = &pWindowRes->resultInfo[i];
SResultRowCellInfo *pResultInfo = &pWindowRes->pCellInfo[i];
char * s = getPosInResultPage(pRuntimeEnv, i, pWindowRes, page);
size_t size = pRuntimeEnv->pQuery->pSelectExpr[i].bytes;
@ -269,7 +255,7 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
* since the attribute of "Pos" is bound to each window result when the window result is created in the
* disk-based result buffer.
*/
void copyTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *dst, const SWindowResult *src) {
void copyResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *dst, const SResultRow *src) {
dst->numOfRows = src->numOfRows;
dst->win = src->win;
dst->closed = src->closed;
@ -277,30 +263,35 @@ void copyTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *dst, con
int32_t nOutputCols = pRuntimeEnv->pQuery->numOfOutput;
for (int32_t i = 0; i < nOutputCols; ++i) {
SResultInfo *pDst = &dst->resultInfo[i];
SResultInfo *pSrc = &src->resultInfo[i];
SResultRowCellInfo *pDst = getResultCell(pRuntimeEnv, dst, i);
SResultRowCellInfo *pSrc = getResultCell(pRuntimeEnv, src, i);
char *buf = pDst->interResultBuf;
memcpy(pDst, pSrc, sizeof(SResultInfo));
pDst->interResultBuf = buf; // restore the allocated buffer
// char *buf = pDst->interResultBuf;
memcpy(pDst, pSrc, sizeof(SResultRowCellInfo) + pRuntimeEnv->pCtx[i].interBufBytes);
// pDst->interResultBuf = buf; // restore the allocated buffer
// copy the result info struct
memcpy(pDst->interResultBuf, pSrc->interResultBuf, pDst->bufLen);
// memcpy(pDst->interResultBuf, pSrc->interResultBuf, pRuntimeEnv->pCtx[i].interBufBytes);
// copy the output buffer data from src to dst, the position info keep unchanged
tFilePage *dstpage = getResBufPage(pRuntimeEnv->pResultBuf, dst->pageId);
char * dstBuf = getPosInResultPage(pRuntimeEnv, i, dst, dstpage);
tFilePage *srcpage = getResBufPage(pRuntimeEnv->pResultBuf, src->pageId);
char * srcBuf = getPosInResultPage(pRuntimeEnv, i, (SWindowResult *)src, srcpage);
char * srcBuf = getPosInResultPage(pRuntimeEnv, i, (SResultRow *)src, srcpage);
size_t s = pRuntimeEnv->pQuery->pSelectExpr[i].bytes;
memcpy(dstBuf, srcBuf, s);
}
}
SResultRowCellInfo* getResultCell(SQueryRuntimeEnv* pRuntimeEnv, const SResultRow* pRow, int32_t index) {
assert(index >= 0 && index < pRuntimeEnv->pQuery->numOfOutput);
return (SResultRowCellInfo*)((char*) pRow->pCellInfo + pRuntimeEnv->rowCellInfoOffset[index]);
}
size_t getWindowResultSize(SQueryRuntimeEnv* pRuntimeEnv) {
return (pRuntimeEnv->pQuery->numOfOutput * sizeof(SResultInfo)) + pRuntimeEnv->interBufSize + sizeof(SWindowResult);
return (pRuntimeEnv->pQuery->numOfOutput * sizeof(SResultRowCellInfo)) + pRuntimeEnv->interBufSize + sizeof(SResultRow);
}
SWindowResultPool* initWindowResultPool(size_t size) {
@ -320,7 +311,7 @@ SWindowResultPool* initWindowResultPool(size_t size) {
return p;
}
SWindowResult* getNewWindowResult(SWindowResultPool* p) {
SResultRow* getNewWindowResult(SWindowResultPool* p) {
if (p == NULL) {
return NULL;
}

9
src/util/inc/hash.h
View File

@ -31,13 +31,16 @@ extern "C" {
typedef void (*_hash_free_fn_t)(void *param);
typedef struct SHashNode {
char *key;
// char *key;
struct SHashNode *next;
uint32_t hashVal; // the hash value of key, if hashVal == HASH_VALUE_IN_TRASH, this node is moved to trash
uint32_t hashVal; // the hash value of key
uint32_t keyLen; // length of the key
char *data;
// char *data;
} SHashNode;
#define GET_HASH_NODE_KEY(_n) ((char*)(_n) + sizeof(SHashNode))
#define GET_HASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHashNode) + (_n)->keyLen)
typedef enum SHashLockTypeE {
HASH_NO_LOCK = 0,
HASH_ENTRY_LOCK = 1,

49
src/util/src/hash.c
View File

@ -22,14 +22,13 @@
#define DO_FREE_HASH_NODE(_n) \
do { \
taosTFree((_n)->data); \
taosTFree(_n); \
} while (0)
#define FREE_HASH_NODE(_h, _n) \
do { \
if ((_h)->freeFp) { \
(_h)->freeFp((_n)->data); \
(_h)->freeFp(GET_HASH_NODE_DATA(_n)); \
} \
\
DO_FREE_HASH_NODE(_n); \
@ -77,7 +76,7 @@ static FORCE_INLINE int32_t taosHashCapacity(int32_t length) {
static FORCE_INLINE SHashNode *doSearchInEntryList(SHashEntry *pe, const void *key, size_t keyLen, uint32_t hashVal) {
SHashNode *pNode = pe->next;
while (pNode) {
if ((pNode->keyLen == keyLen) && (memcmp(pNode->key, key, keyLen) == 0)) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
assert(pNode->hashVal == hashVal);
break;
}
@ -115,11 +114,13 @@ static SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *p
* @param dsize size of actual data
* @return hash node
*/
static FORCE_INLINE SHashNode *doUpdateHashNode(SHashNode *pNode, SHashNode *pNewNode) {
static FORCE_INLINE SHashNode *doUpdateHashNode(SHashNode* prev, SHashNode *pNode, SHashNode *pNewNode) {
assert(pNode->keyLen == pNewNode->keyLen);
SWAP(pNode->key, pNewNode->key, void *);
SWAP(pNode->data, pNewNode->data, void *);
if (prev != NULL) {
prev->next = pNewNode;
}
pNewNode->next = pNode->next;
return pNewNode;
}
@ -208,12 +209,14 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
assert(pNode == NULL);
}
SHashNode* prev = NULL;
while (pNode) {
if ((pNode->keyLen == keyLen) && (memcmp(pNode->key, key, keyLen) == 0)) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
assert(pNode->hashVal == hashVal);
break;
}
prev = pNode;
pNode = pNode->next;
}
@ -239,7 +242,7 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
} else {
// not support the update operation, return error
if (pHashObj->enableUpdate) {
doUpdateHashNode(pNode, pNewNode);
doUpdateHashNode(prev, pNode, pNewNode);
}
if (pHashObj->type == HASH_ENTRY_LOCK) {
@ -293,13 +296,13 @@ void* taosHashGetCB(SHashObj *pHashObj, const void *key, size_t keyLen, void (*f
SHashNode *pNode = doSearchInEntryList(pe, key, keyLen, hashVal);
if (pNode != NULL) {
if (fp != NULL) {
fp(pNode->data);
fp(GET_HASH_NODE_DATA(pNode));
}
if (d != NULL) {
memcpy(d, pNode->data, dsize);
memcpy(d, GET_HASH_NODE_DATA(pNode), dsize);
} else {
data = pNode->data;
data = GET_HASH_NODE_DATA(pNode);
}
}
@ -357,13 +360,13 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
SHashNode *pRes = NULL;
// remove it
if ((pNode->keyLen == keyLen) && (memcmp(pNode->key, key, keyLen) == 0)) {
if ((pNode->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY(pNode), key, keyLen) == 0)) {
pe->num -= 1;
pRes = pNode;
pe->next = pNode->next;
} else {
while (pNode->next != NULL) {
if (((pNode->next)->keyLen == keyLen) && (memcmp((pNode->next)->key, key, keyLen) == 0)) {
if (((pNode->next)->keyLen == keyLen) && (memcmp(GET_HASH_NODE_KEY((pNode->next)), key, keyLen) == 0)) {
assert((pNode->next)->hashVal == hashVal);
break;
}
@ -392,7 +395,7 @@ int32_t taosHashRemoveWithData(SHashObj *pHashObj, const void *key, size_t keyLe
__rd_unlock(&pHashObj->lock, pHashObj->type);
if (data != NULL && pRes != NULL) {
memcpy(data, pRes->data, dsize);
memcpy(data, GET_HASH_NODE_DATA(pRes), dsize);
}
if (pRes != NULL) {
@ -426,7 +429,7 @@ int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), voi
// todo remove the first node
SHashNode *pNode = NULL;
while((pNode = pEntry->next) != NULL) {
if (fp && (!fp(param, pNode->data))) {
if (fp && (!fp(param, GET_HASH_NODE_DATA(pNode)))) {
pEntry->num -= 1;
atomic_sub_fetch_64(&pHashObj->size, 1);
@ -451,7 +454,7 @@ int32_t taosHashCondTraverse(SHashObj *pHashObj, bool (*fp)(void *, void *), voi
while ((pNext = pNode->next) != NULL) {
// not qualified, remove it
if (fp && (!fp(param, pNext->data))) {
if (fp && (!fp(param, GET_HASH_NODE_DATA(pNext)))) {
pNode->next = pNext->next;
pEntry->num -= 1;
atomic_sub_fetch_64(&pHashObj->size, 1);
@ -605,7 +608,7 @@ bool taosHashIterNext(SHashMutableIterator *pIter) {
}
}
void *taosHashIterGet(SHashMutableIterator *iter) { return (iter == NULL) ? NULL : iter->pCur->data; }
void *taosHashIterGet(SHashMutableIterator *iter) { return (iter == NULL) ? NULL : GET_HASH_NODE_DATA(iter->pCur); }
void *taosHashDestroyIter(SHashMutableIterator *iter) {
if (iter == NULL) {
@ -743,21 +746,19 @@ void taosHashTableResize(SHashObj *pHashObj) {
}
SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, size_t dsize, uint32_t hashVal) {
SHashNode *pNewNode = calloc(1, sizeof(SHashNode));
SHashNode *pNewNode = calloc(1, sizeof(SHashNode) + keyLen + dsize);
if (pNewNode == NULL) {
uError("failed to allocate memory, reason:%s", strerror(errno));
return NULL;
}
pNewNode->data = malloc(dsize + keyLen);
memcpy(pNewNode->data, pData, dsize);
pNewNode->key = pNewNode->data + dsize;
memcpy(pNewNode->key, key, keyLen);
pNewNode->keyLen = (uint32_t)keyLen;
pNewNode->hashVal = hashVal;
memcpy(GET_HASH_NODE_DATA(pNewNode), pData, dsize);
memcpy(GET_HASH_NODE_KEY(pNewNode), key, keyLen);
return pNewNode;
}