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

enh: interp fill linear ignore null values during calculation

This commit is contained in:
Ganlin Zhao 2022-11-08 14:50:03 +08:00
parent a7d21c9ee7
commit 6aca4b8a5c
2 changed files with 25 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/executor/inc/tfill.h
View File

@ -36,7 +36,6 @@ typedef struct SFillColInfo {
typedef struct SFillLinearInfo { typedef struct SFillLinearInfo {
SPoint start; SPoint start;
SPoint end; SPoint end;
bool hasNull;
int16_t type; int16_t type;
int32_t bytes; int32_t bytes;
} SFillLinearInfo; } SFillLinearInfo;

43
source/libs/executor/src/timewindowoperator.c
View File

@ -1999,7 +1999,7 @@ static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlo
SFillLinearInfo* pLinearInfo = taosArrayGet(pSliceInfo->pLinearInfo, i); SFillLinearInfo* pLinearInfo = taosArrayGet(pSliceInfo->pLinearInfo, i);
// null data should not be kept since it can not be used to perform interpolation // null data should not be kept since it can not be used to perform interpolation
if (!colDataIsNull_s(pColInfoData, i)) { if (!colDataIsNull_s(pColInfoData, rowIndex)) {
if (pLinearInfo->start.key == INT64_MIN) { if (pLinearInfo->start.key == INT64_MIN) {
pLinearInfo->start.key = *(int64_t*)colDataGetData(pTsCol, rowIndex); pLinearInfo->start.key = *(int64_t*)colDataGetData(pTsCol, rowIndex);
memcpy(pLinearInfo->start.val, colDataGetData(pColInfoData, rowIndex), pLinearInfo->bytes); memcpy(pLinearInfo->start.val, colDataGetData(pColInfoData, rowIndex), pLinearInfo->bytes);
@ -2017,7 +2017,8 @@ static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlo
} }
static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pResBlock) { static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pResBlock,
SSDataBlock* pSrcBlock, int32_t index, bool beforeRange) {
int32_t rows = pResBlock->info.rows; int32_t rows = pResBlock->info.rows;
blockDataEnsureCapacity(pResBlock, rows + 1); blockDataEnsureCapacity(pResBlock, rows + 1);
// todo set the correct primary timestamp column // todo set the correct primary timestamp column
@ -2036,7 +2037,6 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
} }
int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId; int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
// SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
switch (pSliceInfo->fillType) { switch (pSliceInfo->fillType) {
case TSDB_FILL_NULL: { case TSDB_FILL_NULL: {
colDataAppendNULL(pDst, rows); colDataAppendNULL(pDst, rows);
@ -2071,17 +2071,17 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
current.val = taosMemoryCalloc(pLinearInfo->bytes, 1); current.val = taosMemoryCalloc(pLinearInfo->bytes, 1);
// before interp range, do not fill // before interp range, do not fill
if (start.key == INT64_MIN || end.key == INT64_MAX) { SColumnInfoData* pSrc = taosArrayGet(pSrcBlock->pDataBlock, srcSlot);
if (colDataIsNull_s(pSrc, index) || (!beforeRange && end.key == INT64_MAX)) {
hasInterp = false; hasInterp = false;
break; break;
} else if (start.key == INT64_MIN || (beforeRange && end.key == INT64_MAX)) {
//hasInterp = true;
return true;
} }
if (pLinearInfo->hasNull) { taosGetLinearInterpolationVal(&current, pLinearInfo->type, &start, &end, pLinearInfo->type);
colDataAppendNULL(pDst, rows); colDataAppend(pDst, rows, (char*)current.val, false);
} else {
taosGetLinearInterpolationVal(&current, pLinearInfo->type, &start, &end, pLinearInfo->type);
colDataAppend(pDst, rows, (char*)current.val, false);
}
taosMemoryFree(current.val); taosMemoryFree(current.val);
break; break;
@ -2117,6 +2117,8 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
if (hasInterp) { if (hasInterp) {
pResBlock->info.rows += 1; pResBlock->info.rows += 1;
} }
return hasInterp;
} }
static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) { static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
@ -2192,7 +2194,6 @@ static int32_t initFillLinearInfo(STimeSliceOperatorInfo* pInfo, SSDataBlock* pB
linearInfo.end.key = INT64_MAX; linearInfo.end.key = INT64_MAX;
linearInfo.start.val = taosMemoryCalloc(1, pColInfo->info.bytes); linearInfo.start.val = taosMemoryCalloc(1, pColInfo->info.bytes);
linearInfo.end.val = taosMemoryCalloc(1, pColInfo->info.bytes); linearInfo.end.val = taosMemoryCalloc(1, pColInfo->info.bytes);
linearInfo.hasNull = false;
linearInfo.type = pColInfo->info.type; linearInfo.type = pColInfo->info.type;
linearInfo.bytes = pColInfo->info.bytes; linearInfo.bytes = pColInfo->info.bytes;
taosArrayPush(pInfo->pLinearInfo, &linearInfo); taosArrayPush(pInfo->pLinearInfo, &linearInfo);
@ -2306,9 +2307,12 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1); int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
if (nextTs > pSliceInfo->current) { if (nextTs > pSliceInfo->current) {
while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) { while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); if (!genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock, pBlock, i, false) && pSliceInfo->fillType == TSDB_FILL_LINEAR) {
pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, break;
pInterval->precision); } else {
pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
}
} }
if (pSliceInfo->current > pSliceInfo->win.ekey) { if (pSliceInfo->current > pSliceInfo->win.ekey) {
@ -2327,9 +2331,12 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
doKeepLinearInfo(pSliceInfo, pBlock, i); doKeepLinearInfo(pSliceInfo, pBlock, i);
while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) { while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); if (!genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock, pBlock, i, true) && pSliceInfo->fillType == TSDB_FILL_LINEAR) {
pSliceInfo->current = break;
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); } else {
pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
}
} }
// add current row if timestamp match // add current row if timestamp match
@ -2376,7 +2383,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
// except for fill(next), fill(linear) // except for fill(next), fill(linear)
while (pSliceInfo->current <= pSliceInfo->win.ekey && pSliceInfo->fillType != TSDB_FILL_NEXT && while (pSliceInfo->current <= pSliceInfo->win.ekey && pSliceInfo->fillType != TSDB_FILL_NEXT &&
pSliceInfo->fillType != TSDB_FILL_LINEAR) { pSliceInfo->fillType != TSDB_FILL_LINEAR) {
genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock, NULL, 0, false);
pSliceInfo->current = pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
} }