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

Merge pull request #16998 from taosdata/feature/TD-18820 

fix(stream): data block capacity
This commit is contained in:
Shengliang Guan 2022-09-22 16:38:13 +08:00 committed by GitHub
parent bc1e1f1bec 3a21da6e46
commit 570730d495
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 24 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
source/common/src/tdatablock.c
View File

@ -1278,7 +1278,9 @@ int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
colDataAssign(pDst, pSrc, src->info.rows, &src->info);
}
uint32_t cap = dst->info.capacity;
dst->info = src->info;
dst->info.capacity = cap;
return 0;
}
@ -1302,8 +1304,9 @@ int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
colDataAssign(pDst, pSrc, src->info.rows, &src->info);
}
uint32_t cap = dst->info.capacity;
dst->info = src->info;
dst->info.capacity = cap;
return TSDB_CODE_SUCCESS;
}
@ -1333,6 +1336,8 @@ SSDataBlock* createSpecialDataBlock(EStreamType type) {
// group id
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
infoData.info.bytes = sizeof(TSKEY);
// calculate start ts
taosArrayPush(pBlock->pDataBlock, &infoData);
// calculate end ts
@ -2280,4 +2285,3 @@ const char* blockDecode(SSDataBlock* pBlock, const char* pData) {
ASSERT(pStart - pData == dataLen);
return pStart;
}

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

@ -1303,10 +1303,14 @@ void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t*
SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
SColumnInfoData* pCalStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
SColumnInfoData* pCalEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
colDataAppend(pStartTsCol, pBlock->info.rows, (const char*)pStartTs, false);
colDataAppend(pEndTsCol, pBlock->info.rows, (const char*)pEndTs, false);
colDataAppend(pUidCol, pBlock->info.rows, (const char*)pUid, false);
colDataAppend(pGpCol, pBlock->info.rows, (const char*)pGp, false);
colDataAppendNULL(pCalStartCol, pBlock->info.rows);
colDataAppendNULL(pCalEndCol, pBlock->info.rows);
pBlock->info.rows++;
}

26
source/libs/function/src/builtinsimpl.c
View File

@ -47,6 +47,7 @@ typedef struct SSumRes {
uint64_t usum;
double dsum;
};
int16_t type;
} SSumRes;
typedef struct SAvgRes {
@ -73,6 +74,7 @@ typedef struct SMinmaxResInfo {
STuplePos nullTuplePos;
bool nullTupleSaved;
int16_t type;
} SMinmaxResInfo;
typedef struct STopBotResItem {
@ -485,8 +487,7 @@ int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SFirstLastRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
int32_t bytes = pDestCtx->input.pData[0]->info.bytes;
int32_t bytes = pDBuf->bytes;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SFirstLastRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
@ -617,6 +618,7 @@ int32_t sumFunction(SqlFunctionCtx* pCtx) {
int32_t type = pInput->pData[0]->info.type;
SSumRes* pSumRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
pSumRes->type = type;
if (IS_NULL_TYPE(type)) {
numOfElem = 0;
@ -740,10 +742,10 @@ int32_t sumInvertFunction(SqlFunctionCtx* pCtx) {
int32_t sumCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SSumRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SSumRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
int16_t type = pDBuf->type == TSDB_DATA_TYPE_NULL ? pSBuf->type : pDBuf->type;
if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) {
pDBuf->isum += pSBuf->isum;
@ -1072,10 +1074,10 @@ int32_t avgInvertFunction(SqlFunctionCtx* pCtx) {
int32_t avgCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SAvgRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SAvgRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
int16_t type = pDBuf->type == TSDB_DATA_TYPE_NULL ? pSBuf->type : pDBuf->type;
if (IS_SIGNED_NUMERIC_TYPE(type)) {
pDBuf->sum.isum += pSBuf->sum.isum;
@ -1181,6 +1183,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SMinmaxResInfo* pBuf = GET_ROWCELL_INTERBUF(pResInfo);
pBuf->type = type;
if (IS_NULL_TYPE(type)) {
numOfElems = 0;
@ -1729,10 +1732,10 @@ void replaceTupleData(STuplePos* pDestPos, STuplePos* pSourcePos) {
int32_t minMaxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t isMinFunc) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SMinmaxResInfo* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SMinmaxResInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
int16_t type = pDBuf->type == TSDB_DATA_TYPE_NULL ? pSBuf->type : pDBuf->type;
if (IS_FLOAT_TYPE(type)) {
if (pSBuf->assign && ((((*(double*)&pDBuf->v) < (*(double*)&pSBuf->v)) ^ isMinFunc) || !pDBuf->assign)) {
*(double*)&pDBuf->v = *(double*)&pSBuf->v;
@ -2105,10 +2108,10 @@ int32_t stddevPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t stddevCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SStddevRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SStddevRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
int16_t type = pDBuf->type == TSDB_DATA_TYPE_NULL ? pSBuf->type : pDBuf->type;
if (IS_SIGNED_NUMERIC_TYPE(type)) {
pDBuf->isum += pSBuf->isum;
@ -3069,8 +3072,7 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SFirstLastRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
int32_t type = pDestCtx->input.pData[0]->info.type;
int32_t bytes = pDestCtx->input.pData[0]->info.bytes;
int32_t bytes = pDBuf->bytes;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SFirstLastRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
@ -3746,9 +3748,9 @@ void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type,
}
int32_t topCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
STopBotRes* pSBuf = getTopBotOutputInfo(pSourceCtx);
int16_t type = pSBuf->type;
for (int32_t i = 0; i < pSResInfo->numOfRes; i++) {
addResult(pDestCtx, pSBuf->pItems + i, type, true);
}
@ -3756,9 +3758,9 @@ int32_t topCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
}
int32_t bottomCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
int32_t type = pDestCtx->input.pData[0]->info.type;
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
STopBotRes* pSBuf = getTopBotOutputInfo(pSourceCtx);
int16_t type = pSBuf->type;
for (int32_t i = 0; i < pSResInfo->numOfRes; i++) {
addResult(pDestCtx, pSBuf->pItems + i, type, false);
}
@ -5414,8 +5416,8 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
int32_t i = pInput->startRowIndex;
if (pCtx->start.key != INT64_MIN) {
//ASSERT((pCtx->start.key < tsList[i] && pCtx->order == TSDB_ORDER_ASC) ||
// (pCtx->start.key > tsList[i] && pCtx->order == TSDB_ORDER_DESC));
// ASSERT((pCtx->start.key < tsList[i] && pCtx->order == TSDB_ORDER_ASC) ||
// (pCtx->start.key > tsList[i] && pCtx->order == TSDB_ORDER_DESC));
ASSERT(last->key == INT64_MIN);
for (; i < pInput->numOfRows + pInput->startRowIndex; ++i) {