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

fix(query): set correct ts column index in interp executor.

This commit is contained in:
Haojun Liao 2022-06-21 09:41:35 +08:00
parent 1bf7fbbc0b
commit e5d25f4634
3 changed files with 63 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
source/client/test/clientTests.cpp
View File

@ -43,6 +43,26 @@ void showDB(TAOS* pConn) {
}
}
void printResult(TAOS_RES* pRes) {
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
int32_t n = 0;
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t* length = taos_fetch_lengths(pRes);
for(int32_t i = 0; i < numOfFields; ++i) {
printf("(%d):%d " , i, length[i]);
}
printf("\n");
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
memset(str, 0, sizeof(str));
}
}
void fetchCallback(void* param, void* res, int32_t numOfRow) {
#if 0
printf("numOfRow = %d \n", numOfRow);
@ -729,48 +749,31 @@ TEST(testCase, projection_query_tables) {
// taos_close(pConn);
//}
//TEST(testCase, agg_query_tables) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// ASSERT_NE(pConn, nullptr);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("failed to use db, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "show stables");
// if (taos_errno(pRes) != 0) {
// printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
//
// TAOS_ROW pRow = NULL;
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// int32_t n = 0;
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t* length = taos_fetch_lengths(pRes);
// for(int32_t i = 0; i < numOfFields; ++i) {
// printf("(%d):%d " , i, length[i]);
// }
// printf("\n");
//
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// memset(str, 0, sizeof(str));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
#endif
TEST(testCase, agg_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("failed to use db, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
taos_free_result(pRes);
pRes = taos_query(pConn, "show table distributed st1");
if (taos_errno(pRes) != 0) {
printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
printResult(pRes);
taos_free_result(pRes);
taos_close(pConn);
}
#endif
/*
--- copy the following script in the shell to setup the environment ---
@ -786,7 +789,7 @@ TEST(testCase, async_api_test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
taos_query(pConn, "use table_alltype_hyperloglog");
taos_query(pConn, "use abc1");
#if 0
TAOS_RES* pRes = taos_query(pConn, "insert into tu(ts) values('2022-02-27 12:12:61')");
if (taos_errno(pRes) != 0) {
@ -812,7 +815,7 @@ TEST(testCase, async_api_test) {
}
#endif
taos_query_a(pConn, "select HYPERLOGLOG(q_ts) from stable_1_2 where ts between 1630000001000 and 1630100001000 interval(19d) Fill(NONE);", queryCallback, pConn);
taos_query_a(pConn, "select count(*) from tu", queryCallback, pConn);
getchar();
taos_close(pConn);
}

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

@ -1888,13 +1888,13 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlot);
SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
char* v = colDataGetData(pSrc, i);
colDataAppend(pDst, numOfRows, v, false);
}
numOfRows += 1;
pResBlock->info.rows += 1;
doKeepPrevRows(pSliceInfo, pBlock, i);
pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
@ -1902,15 +1902,21 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
doSetOperatorCompleted(pOperator);
break;
}
if (pResBlock->info.rows >= pResBlock->info.capacity) {
break;
}
} else if (ts < pSliceInfo->current) {
if (i < pBlock->info.rows - 1) {
int64_t nextTs = *(int64_t*) colDataGetData(pTsCol, i + 1);
if (nextTs > pSliceInfo->current) {
while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
numOfRows += 1;
pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
if (pResBlock->info.rows >= pResBlock->info.capacity) {
break;
}
}
if (pSliceInfo->current > pSliceInfo->win.ekey) {
@ -1926,9 +1932,11 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
} else { // ts > pSliceInfo->current
while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
numOfRows += 1;
pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
if (pResBlock->info.rows >= pResBlock->info.capacity) {
break;
}
}
if (pSliceInfo->current > pSliceInfo->win.ekey) {

7
source/libs/planner/src/planPhysiCreater.c
View File

@ -879,12 +879,15 @@ static int32_t createInterpFuncPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pCh
pInterpFunc->interval = pFuncLogicNode->interval;
pInterpFunc->fillMode = pFuncLogicNode->fillMode;
pInterpFunc->pFillValues = nodesCloneNode(pFuncLogicNode->pFillValues);
pInterpFunc->pTimeSeries = nodesCloneNode(pFuncLogicNode->pTimeSeries);
if (NULL == pInterpFunc->pTimeSeries || (NULL != pFuncLogicNode->pFillValues && NULL == pInterpFunc->pFillValues)) {
if (NULL != pFuncLogicNode->pFillValues && NULL == pInterpFunc->pFillValues) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
code = setNodeSlotId(pCxt, pChildTupe->dataBlockId, -1, pFuncLogicNode->pTimeSeries, &pInterpFunc->pTimeSeries);
}
if (TSDB_CODE_SUCCESS == code) {
*pPhyNode = (SPhysiNode*)pInterpFunc;
} else {