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

refactor: do some internal refactor of sample function.

This commit is contained in:
Haojun Liao 2022-06-12 15:25:37 +08:00
parent c2e8aa9659
commit d0dd451494
3 changed files with 25 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/function/inc/builtinsimpl.h
View File

@ -155,6 +155,7 @@ int32_t mavgFunction(SqlFunctionCtx* pCtx);
bool getSampleFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool sampleFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t sampleFunction(SqlFunctionCtx* pCtx);
int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getTailFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool tailFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);

4
source/libs/function/src/builtins.c
View File

@ -1751,12 +1751,12 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "sample",
.type = FUNCTION_TYPE_SAMPLE,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
.translateFunc = translateSample,
.getEnvFunc = getSampleFuncEnv,
.initFunc = sampleFunctionSetup,
.processFunc = sampleFunction,
.finalizeFunc = NULL
.finalizeFunc = sampleFinalize
},
{
.name = "tail",

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

@ -198,7 +198,7 @@ typedef struct SSampleInfo {
int32_t numSampled;
uint8_t colType;
int16_t colBytes;
char *data;
char *data;
int64_t *timestamp;
} SSampleInfo;
@ -4183,28 +4183,38 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
SSampleInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
TSKEY* tsList = NULL;
if (pInput->pPTS != NULL) {
tsList = (int64_t*)pInput->pPTS->pData;
}
SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
int32_t alreadySampled = pInfo->numSampled;
int32_t startOffset = pCtx->offset;
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
if (colDataIsNull_s(pInputCol, i)) {
//colDataAppendNULL(pOutput, i);
continue;
}
char* data = colDataGetData(pInputCol, i);
doReservoirSample(pInfo, data, tsList[i], i);
doReservoirSample(pInfo, data, /*tsList[i]*/0, i);
}
SET_VAL(pResInfo, pInfo->numSampled, pInfo->numSampled);
return TSDB_CODE_SUCCESS;
}
int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
SSampleInfo* pInfo = GET_ROWCELL_INTERBUF(pEntryInfo);
pEntryInfo->complete = true;
int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
int32_t currentRow = pBlock->info.rows;
for (int32_t i = 0; i < pInfo->numSampled; ++i) {
int32_t pos = startOffset + i;
colDataAppend(pOutput, pos, pInfo->data + i * pInfo->colBytes, false);
//TODO: handle ts output
colDataAppend(pCol, currentRow + i, pInfo->data + i * pInfo->colBytes, false);
}
return pInfo->numSampled;