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

Merge remote-tracking branch 'origin/feature/3.0_liaohj' into feature/qnode

This commit is contained in:
dapan1121 2022-01-11 11:20:29 +08:00
parent 89dcffce88 a2545549bd
commit 9f864b7440
29 changed files with 2277 additions and 2003 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/common.h
View File

@ -62,6 +62,12 @@ typedef struct SConstantItem {
SVariant value;
} SConstantItem;
typedef struct {
uint32_t numOfTables;
SArray *pGroupList;
SHashObj *map; // speedup acquire the tableQueryInfo by table uid
} STableGroupInfo;
typedef struct SSDataBlock {
SColumnDataAgg *pBlockAgg;
SArray *pDataBlock; // SArray<SColumnInfoData>

134
include/dnode/vnode/tsdb/tsdb.h
View File

@ -18,6 +18,7 @@
#include "mallocator.h"
#include "meta.h"
#include "common.h"
#ifdef __cplusplus
extern "C" {
@ -39,6 +40,10 @@ typedef struct STable {
STSchema *pSchema;
} STable;
#define BLOCK_LOAD_OFFSET_SEQ_ORDER 1
#define BLOCK_LOAD_TABLE_SEQ_ORDER 2
#define BLOCK_LOAD_TABLE_RR_ORDER 3
#define TABLE_TID(t) (t)->tid
#define TABLE_UID(t) (t)->uid
@ -58,6 +63,22 @@ typedef struct STsdbCfg {
int8_t compression;
} STsdbCfg;
// query condition to build multi-table data block iterator
typedef struct STsdbQueryCond {
STimeWindow twindow;
int32_t order; // desc|asc order to iterate the data block
int32_t numOfCols;
SColumnInfo *colList;
bool loadExternalRows; // load external rows or not
int32_t type; // data block load type:
} STsdbQueryCond;
typedef struct {
void *pTable;
TSKEY lastKey;
uint64_t uid;
} STableKeyInfo;
// STsdb
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta);
void tsdbClose(STsdb *);
@ -70,6 +91,119 @@ int tsdbCommit(STsdb *pTsdb);
int tsdbOptionsInit(STsdbCfg *);
void tsdbOptionsClear(STsdbCfg *);
typedef void* tsdbReadHandleT;
/**
* Get the data block iterator, starting from position according to the query condition
*
* @param tsdb tsdb handle
* @param pCond query condition, including time window, result set order, and basic required columns for each block
* @param tableInfoGroup table object list in the form of set, grouped into different sets according to the
* group by condition
* @param qinfo query info handle from query processor
* @return
*/
tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId,
void *pRef);
/**
* Get the last row of the given query time window for all the tables in STableGroupInfo object.
* Note that only one data block with only row will be returned while invoking retrieve data block function for
* all tables in this group.
*
* @param tsdb tsdb handle
* @param pCond query condition, including time window, result set order, and basic required columns for each block
* @param tableInfo table list.
* @return
*/
//tsdbReadHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
// SMemRef *pRef);
tsdbReadHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, void* pMemRef);
bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle);
/**
* get num of rows in mem table
*
* @param pHandle
* @return row size
*/
int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle);
/**
* move to next block if exists
*
* @param pTsdbReadHandle
* @return
*/
bool tsdbNextDataBlock(tsdbReadHandleT pTsdbReadHandle);
/**
* Get current data block information
*
* @param pTsdbReadHandle
* @param pBlockInfo
* @return
*/
void tsdbRetrieveDataBlockInfo(tsdbReadHandleT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
/**
*
* Get the pre-calculated information w.r.t. current data block.
*
* In case of data block in cache, the pBlockStatis will always be NULL.
* If a block is not completed loaded from disk, the pBlockStatis will be NULL.
* @pBlockStatis the pre-calculated value for current data blocks. if the block is a cache block, always return 0
* @return
*/
int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReadHandleT *pTsdbReadHandle, SDataStatis **pBlockStatis);
/**
*
* The query condition with primary timestamp is passed to iterator during its constructor function,
* the returned data block must be satisfied with the time window condition in any cases,
* which means the SData data block is not actually the completed disk data blocks.
*
* @param pTsdbReadHandle query handle
* @param pColumnIdList required data columns id list
* @return
*/
SArray *tsdbRetrieveDataBlock(tsdbReadHandleT *pTsdbReadHandle, SArray *pColumnIdList);
/**
* destroy the created table group list, which is generated by tag query
* @param pGroupList
*/
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
/**
* create the table group result including only one table, used to handle the normal table query
*
* @param tsdb tsdbHandle
* @param uid table uid
* @param pGroupInfo the generated result
* @return
*/
int32_t tsdbGetOneTableGroup(STsdb *tsdb, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
/**
*
* @param tsdb
* @param pTableIdList
* @param pGroupInfo
* @return
*/
int32_t tsdbGetTableGroupFromIdList(STsdb *tsdb, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
/**
* clean up the query handle
* @param queryHandle
*/
void tsdbCleanupQueryHandle(tsdbReadHandleT queryHandle);
#ifdef __cplusplus
}
#endif

26
include/libs/executor/executor.h
View File

@ -22,23 +22,25 @@ extern "C" {
typedef void* qTaskInfo_t;
/**
* create the qinfo object according to QueryTableMsg
* @param tsdb
* @param pQueryTableMsg
* @param pTaskInfo
* @return
*/
int32_t qCreateTask(void* tsdb, int32_t vgId, void* pQueryTableMsg, qTaskInfo_t* pTaskInfo, uint64_t qId);
/**
* Create the exec task object according to task json
* @param tsdb
* @param vgId
* @param pTaskInfoMsg
* @param pTaskInfo
* @param qId
* @return
*/
int32_t qCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo);
/**
* the main query execution function, including query on both table and multiple tables,
* the main task execution function, including query on both table and multiple tables,
* which are decided according to the tag or table name query conditions
*
* @param qinfo
* @return
*/
bool qExecTask(qTaskInfo_t qinfo, uint64_t *qId);
bool qExecTask(qTaskInfo_t qTask, SSDataBlock** pRes);
/**
* Retrieve the produced results information, if current query is not paused or completed,
@ -81,7 +83,7 @@ int32_t qKillTask(qTaskInfo_t qinfo);
* @param qinfo
* @return
*/
int32_t qIsQueryCompleted(qTaskInfo_t qinfo);
int32_t qIsTaskCompleted(qTaskInfo_t qinfo);
/**
* destroy query info structure
@ -113,7 +115,7 @@ int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t
* @param numOfIndex
* @return
*/
int32_t qCreateTableGroupByGroupExpr(SArray* pTableIdList, TSKEY skey, STableGroupInfo groupInfo, SColIndex* groupByIndex, int32_t numOfIndex);
//int32_t qCreateTableGroupByGroupExpr(SArray* pTableIdList, TSKEY skey, STableGroupInfo groupInfo, SColIndex* groupByIndex, int32_t numOfIndex);
/**
* Update the table id list of a given query.

2
include/libs/planner/planner.h
View File

@ -150,7 +150,7 @@ struct SQueryNode;
* @param requestId
* @return
*/
int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag** pDag, uint64_t requestId);
int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag** pDag, SSchema** pSchema, uint32_t* numOfResCols, uint64_t requestId);
// Set datasource of this subplan, multiple calls may be made to a subplan.
// @subplan subplan to be schedule

20
source/client/src/clientImpl.c
View File

@ -197,7 +197,25 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag) {
pRequest->type = pQueryNode->type;
return qCreateQueryDag(pQueryNode, pDag, pRequest->requestId);
SSchema *pSchema = NULL;
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
int32_t code = qCreateQueryDag(pQueryNode, pDag, &pSchema, &pResInfo->numOfCols, pRequest->requestId);
if (code != 0) {
return code;
}
if (pQueryNode->type == TSDB_SQL_SELECT) {
pResInfo->fields = calloc(1, sizeof(TAOS_FIELD));
for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
pResInfo->fields[i].bytes = pSchema[i].bytes;
pResInfo->fields[i].type = pSchema[i].type;
tstrncpy(pResInfo->fields[i].name, pSchema[i].name, tListLen(pResInfo->fields[i].name));
}
}
return code;
}
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, void** pJob) {

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

@ -24,7 +24,6 @@
#include "../inc/clientInt.h"
#include "taos.h"
#include "tglobal.h"
namespace {
void showDB(TAOS* pConn) {
@ -57,449 +56,449 @@ TEST(testCase, connect_Test) {
taos_close(pConn);
}
TEST(testCase, create_user_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create user abc pass 'abc'");
if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
printf("failed to create user, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, create_account_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create account aabc pass 'abc'");
if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
printf("failed to create user, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, drop_account_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "drop account aabc");
if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
printf("failed to create user, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, show_user_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "show users");
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, drop_user_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "drop user abc");
if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
printf("failed to create user, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, show_db_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "show databases");
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_close(pConn);
}
TEST(testCase, create_db_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
if (taos_errno(pRes) != 0) {
printf("error in create db, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_free_result(pRes);
pRes = taos_query(pConn, "create database abc1 vgroups 4");
if (taos_errno(pRes) != 0) {
printf("error in create db, reason:%s\n", taos_errstr(pRes));
}
taos_close(pConn);
}
TEST(testCase, create_dnode_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create dnode abc1 port 7000");
if (taos_errno(pRes) != 0) {
printf("error in create dnode, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create dnode 1.1.1.1 port 9000");
if (taos_errno(pRes) != 0) {
printf("failed to create dnode, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, drop_dnode_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "drop dnode 2");
if (taos_errno(pRes) != 0) {
printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, use_db_test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("error in use db, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_close(pConn);
}
// TEST(testCase, drop_db_test) {
//TEST(testCase, create_user_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// showDB(pConn);
//
// TAOS_RES* pRes = taos_query(pConn, "drop database abc1");
// if (taos_errno(pRes) != 0) {
// printf("failed to drop db, reason:%s\n", taos_errstr(pRes));
// TAOS_RES* pRes = taos_query(pConn, "create user abc pass 'abc'");
// if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
// printf("failed to create user, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// showDB(pConn);
//
// pRes = taos_query(pConn, "create database abc1");
// if (taos_errno(pRes) != 0) {
// printf("create to drop db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
// taos_close(pConn);
//}
TEST(testCase, create_stable_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
if (taos_errno(pRes) != 0) {
printf("error in create db, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("error in use db, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
if (taos_errno(pRes) != 0) {
printf("error in create stable, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, create_table_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "create table tm0(ts timestamp, k int)");
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, create_ctable_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
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);
pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
if (taos_errno(pRes) != 0) {
printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, show_stable_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
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);
pRes = taos_query(pConn, "show stables");
if (taos_errno(pRes) != 0) {
printf("failed to show stables, 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);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, show_vgroup_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
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);
pRes = taos_query(pConn, "show vgroups");
if (taos_errno(pRes) != 0) {
printf("failed to show vgroups, 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);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, create_multiple_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", taos_errstr(pRes));
taos_free_result(pRes);
taos_close(pConn);
return;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table t_2 using st1 tags(1)");
if (taos_errno(pRes) != 0) {
printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table t_3 using st1 tags(2)");
if (taos_errno(pRes) != 0) {
printf("failed to create multiple tables, 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);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
for (int32_t i = 0; i < 20; ++i) {
char sql[512] = {0};
snprintf(sql, tListLen(sql),
"create table t_x_%d using st1 tags(2) t_x_%d using st1 tags(5) t_x_%d using st1 tags(911)", i,
(i + 1) * 30, (i + 2) * 40);
TAOS_RES* pres = taos_query(pConn, sql);
if (taos_errno(pres) != 0) {
printf("failed to create table %d\n, reason:%s", i, taos_errstr(pres));
}
taos_free_result(pres);
}
taos_close(pConn);
}
TEST(testCase, show_table_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "show tables");
if (taos_errno(pRes) != 0) {
printf("failed to show vgroups, 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);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, drop_stable_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create database abc1");
if (taos_errno(pRes) != 0) {
printf("error in creating db, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("error in using db, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "drop stable st1");
if (taos_errno(pRes) != 0) {
printf("failed to drop stable, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, generated_request_id_test) {
SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
for (int32_t i = 0; i < 50000; ++i) {
uint64_t v = generateRequestId();
void* result = taosHashGet(phash, &v, sizeof(v));
if (result != nullptr) {
printf("0x%lx, index:%d\n", v, i);
}
assert(result == nullptr);
taosHashPut(phash, &v, sizeof(v), NULL, 0);
}
taosHashCleanup(phash);
}
//
//TEST(testCase, create_account_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create account aabc pass 'abc'");
// if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
// printf("failed to create user, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, drop_account_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "drop account aabc");
// if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
// printf("failed to create user, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_user_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "show users");
// TAOS_ROW pRow = NULL;
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, drop_user_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "drop user abc");
// if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
// printf("failed to create user, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_db_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "show databases");
// TAOS_ROW pRow = NULL;
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_close(pConn);
//}
//
//TEST(testCase, create_db_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create database abc1 vgroups 4");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
// taos_close(pConn);
//}
//
//TEST(testCase, create_dnode_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create dnode abc1 port 7000");
// if (taos_errno(pRes) != 0) {
// printf("error in create dnode, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create dnode 1.1.1.1 port 9000");
// if (taos_errno(pRes) != 0) {
// printf("failed to create dnode, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// taos_close(pConn);
//}
//
//TEST(testCase, drop_dnode_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "drop dnode 2");
// if (taos_errno(pRes) != 0) {
// printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, use_db_test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in use db, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_close(pConn);
//}
//
//// TEST(testCase, drop_db_test) {
//// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
//// assert(pConn != NULL);
////
//// showDB(pConn);
////
//// TAOS_RES* pRes = taos_query(pConn, "drop database abc1");
//// if (taos_errno(pRes) != 0) {
//// printf("failed to drop db, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
////
//// showDB(pConn);
////
//// pRes = taos_query(pConn, "create database abc1");
//// if (taos_errno(pRes) != 0) {
//// printf("create to drop db, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
//// taos_close(pConn);
////}
//
//TEST(testCase, create_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in use db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
// if (taos_errno(pRes) != 0) {
// printf("error in create stable, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, create_table_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create table tm0(ts timestamp, k int)");
// taos_free_result(pRes);
//
// taos_close(pConn);
//}
//
//TEST(testCase, create_ctable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// 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);
//
// pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// 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);
//
// pRes = taos_query(pConn, "show stables");
// if (taos_errno(pRes) != 0) {
// printf("failed to show stables, 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);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_vgroup_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// 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);
//
// pRes = taos_query(pConn, "show vgroups");
// if (taos_errno(pRes) != 0) {
// printf("failed to show vgroups, 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);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, create_multiple_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);
// taos_close(pConn);
// return;
// }
//
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create table t_2 using st1 tags(1)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
//
// taos_free_result(pRes);
// pRes = taos_query(pConn, "create table t_3 using st1 tags(2)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create multiple tables, 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);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
//
// for (int32_t i = 0; i < 20; ++i) {
// char sql[512] = {0};
// snprintf(sql, tListLen(sql),
// "create table t_x_%d using st1 tags(2) t_x_%d using st1 tags(5) t_x_%d using st1 tags(911)", i,
// (i + 1) * 30, (i + 2) * 40);
// TAOS_RES* pres = taos_query(pConn, sql);
// if (taos_errno(pres) != 0) {
// printf("failed to create table %d\n, reason:%s", i, taos_errstr(pres));
// }
// taos_free_result(pres);
// }
//
// taos_close(pConn);
//}
//
//TEST(testCase, show_table_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "show tables");
// if (taos_errno(pRes) != 0) {
// printf("failed to show vgroups, 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);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, drop_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in creating db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in using db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "drop stable st1");
// if (taos_errno(pRes) != 0) {
// printf("failed to drop stable, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, generated_request_id_test) {
// SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
//
// for (int32_t i = 0; i < 50000; ++i) {
// uint64_t v = generateRequestId();
// void* result = taosHashGet(phash, &v, sizeof(v));
// if (result != nullptr) {
// printf("0x%lx, index:%d\n", v, i);
// }
// assert(result == nullptr);
// taosHashPut(phash, &v, sizeof(v), NULL, 0);
// }
//
// taosHashCleanup(phash);
//}
// TEST(testCase, create_topic_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@ -557,18 +556,29 @@ TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use test1");
if (taos_errno(pRes) != 0) {
printf("failed to use db, reason:%s", taos_errstr(pRes));
taos_free_result(pRes);
return;
}
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
// if (taos_errno(pRes) != 0) {
// printf("failed to use db, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// return;
// }
taos_free_result(pRes);
// taos_free_result(pRes);
pRes = taos_query(pConn, "select * from tm0");
TAOS_RES* pRes = taos_query(pConn, "use abc1");
// pRes = taos_query(pConn, "create table m1 (ts timestamp, k int) tags(a int)");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create table tu using m1 tags(1)");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "insert into tu values(now, 1)");
// taos_free_result(pRes);
pRes = taos_query(pConn, "select * from tu");
if (taos_errno(pRes) != 0) {
printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}

2
source/dnode/vnode/impl/src/vnodeQuery.c
View File

@ -23,7 +23,7 @@ int vnodeQueryOpen(SVnode *pVnode) { return qWorkerInit(NULL, &pVnode->pQuery);
int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
vTrace("query message is processed");
return qWorkerProcessQueryMsg(pVnode, pVnode->pQuery, pMsg);
return qWorkerProcessQueryMsg(pVnode->pTsdb, pVnode->pQuery, pMsg);
}
int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {

1
source/dnode/vnode/tsdb/CMakeLists.txt
View File

@ -13,6 +13,7 @@ else(0)
"src/tsdbReadImpl.c"
"src/tsdbFile.c"
"src/tsdbFS.c"
"src/tsdbRead.c"
)
endif(0)

2
source/dnode/vnode/tsdb/src/tsdbCommit.c
View File

@ -1253,7 +1253,7 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDataCols *
pBlock->keyFirst = dataColsKeyFirst(pDataCols);
pBlock->keyLast = dataColsKeyLast(pDataCols);
tsdbDebug("vgId:%d tid:%d a block of data is written to file %s, offset %" PRId64
tsdbDebug("vgId:%d uid:%"PRId64" a block of data is written to file %s, offset %" PRId64
" numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64,
REPO_ID(pRepo), TABLE_TID(pTable), TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len,
pBlock->numOfCols, pBlock->keyFirst, pBlock->keyLast);

2138
source/dnode/vnode/tsdb/src/tsdbRead.c
View File

File diff suppressed because it is too large Load Diff

31
source/libs/executor/CMakeLists.txt
View File

@ -1,12 +1,27 @@
aux_source_directory(src EXECUTOR_SRC)
add_library(executor ${EXECUTOR_SRC})
#add_library(executor ${EXECUTOR_SRC})
#target_link_libraries(
# executor
# PRIVATE os util common function parser planner qcom tsdb
#)
add_library(executor STATIC ${EXECUTOR_SRC})
#set_target_properties(executor PROPERTIES
# IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/libexecutor.a"
# INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_SOURCE_DIR}/include/libs/executor"
# )
target_link_libraries(executor
PRIVATE os util common function parser planner qcom tsdb
)
target_include_directories(
executor
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/executor"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
executor
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/executor"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
executor
PRIVATE os util common function parser planner qcom
)
#if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
#endif(${BUILD_TEST})

2
source/libs/executor/inc/dataSinkMgt.h
View File

@ -33,7 +33,7 @@ struct SDataSink;
struct SSDataBlock;
typedef struct SDataSinkMgtCfg {
uint32_t maxDataBlockNum;
uint32_t maxDataBlockNum; // todo: this should be numOfRows?
uint32_t maxDataBlockNumPerQuery;
} SDataSinkMgtCfg;

4
source/libs/executor/inc/executil.h
View File

@ -38,7 +38,7 @@
#define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
#define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)
#define GET_QID(_r) (((SQInfo*)((_r)->qinfo))->qId)
#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.queryId)
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
@ -157,6 +157,6 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, struct STaskRuntimeEnv *pRuntimeEnv, int32_t* offset);
int32_t initUdfInfo(struct SUdfInfo* pUdfInfo);
//int32_t initUdfInfo(struct SUdfInfo* pUdfInfo);
#endif // TDENGINE_QUERYUTIL_H

50
source/libs/executor/inc/executorimpl.h
View File

@ -29,15 +29,8 @@
#include "tpagedfile.h"
#include "planner.h"
struct SColumnFilterElem;
typedef struct {
uint32_t numOfTables;
SArray *pGroupList;
SHashObj *map; // speedup acquire the tableQueryInfo by table uid
} STableGroupInfo;
typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);
#define IS_QUERY_KILLED(_q) ((_q)->code == TSDB_CODE_TSC_QUERY_CANCELLED)
@ -51,19 +44,19 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
#define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData? 1 : 0)
enum {
// when query starts to execute, this status will set
QUERY_NOT_COMPLETED = 0x1u,
// when this task starts to execute, this status will set
TASK_NOT_COMPLETED = 0x1u,
/* query is over
/* Task is over
* 1. this status is used in one row result query process, e.g., count/sum/first/last/ avg...etc.
* 2. when all data within queried time window, it is also denoted as query_completed
*/
QUERY_COMPLETED = 0x2u,
TASK_COMPLETED = 0x2u,
/* when the result is not completed return to client, this status will be
* usually used in case of interval query with interpolation option
*/
QUERY_OVER = 0x4u,
TASK_OVER = 0x4u,
};
typedef struct SResultRowCell {
@ -129,6 +122,7 @@ typedef struct {
} SOperatorProfResult;
typedef struct STaskCostInfo {
int64_t created;
int64_t start;
int64_t end;
@ -246,13 +240,14 @@ typedef struct STaskIdInfo {
uint64_t taskId; // this is a subplan id
} STaskIdInfo;
typedef struct STaskInfo {
typedef struct SExecTaskInfo {
STaskIdInfo id;
char *content;
uint32_t status;
STimeWindow window;
STaskCostInfo cost;
int64_t owner; // if it is in execution
int32_t code;
STableGroupInfo tableqinfoGroupInfo; // this is a group array list, including SArray<STableQueryInfo*> structure
pthread_mutex_t lock; // used to synchronize the rsp/query threads
@ -260,8 +255,9 @@ typedef struct STaskInfo {
// int32_t dataReady; // denote if query result is ready or not
// void* rspContext; // response context
char *sql; // query sql string
jmp_buf env;
} STaskInfo;
jmp_buf env; //
struct SOperatorInfo *pRoot;
} SExecTaskInfo;
typedef struct STaskRuntimeEnv {
jmp_buf env;
@ -269,7 +265,7 @@ typedef struct STaskRuntimeEnv {
uint32_t status; // query status
void* qinfo;
uint8_t scanFlag; // denotes reversed scan of data or not
void* pQueryHandle;
void* pTsdbReadHandle;
int32_t prevGroupId; // previous executed group id
bool enableGroupData;
@ -314,8 +310,8 @@ typedef struct SOperatorInfo {
char *name; // name, used to show the query execution plan
void *info; // extension attribution
SExprInfo *pExpr;
STaskRuntimeEnv *pRuntimeEnv;
STaskInfo *pTaskInfo;
STaskRuntimeEnv *pRuntimeEnv; // todo remove it
SExecTaskInfo *pTaskInfo;
struct SOperatorInfo **pDownstream; // downstram pointer list
int32_t numOfDownstream; // number of downstream. The value is always ONE expect for join operator
@ -376,7 +372,7 @@ typedef struct STaskParam {
} STaskParam;
typedef struct STableScanInfo {
void *pQueryHandle;
void *pTsdbReadHandle;
int32_t numOfBlocks;
int32_t numOfSkipped;
int32_t numOfBlockStatis;
@ -544,7 +540,7 @@ typedef struct SOrderOperatorInfo {
void appendUpstream(SOperatorInfo* p, SOperatorInfo* pUpstream);
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv, int32_t repeatTime, int32_t reverseTime);
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime);
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv);
SOperatorInfo* createAggregateOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
@ -572,11 +568,11 @@ SOperatorInfo* createFilterOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pUpstream, int32_t numOfUpstream, SSchema* pSchema, int32_t numOfOutput);
SOperatorInfo* createOrderOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrder* pOrderVal);
SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
SSDataBlock* doSLimit(void* param, bool* newgroup);
//SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
//SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
//SSDataBlock* doSLimit(void* param, bool* newgroup);
int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
//int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, SSDataBlock* pBlock);
bool doFilterDataBlock(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, int32_t numOfRows, int8_t* p);
void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p);
@ -617,14 +613,14 @@ STableQueryInfo* createTmpTableQueryInfo(STimeWindow win);
int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
bool isQueryKilled(SQInfo *pQInfo);
bool isTaskKilled(SExecTaskInfo *pTaskInfo);
int32_t checkForQueryBuf(size_t numOfTables);
bool checkNeedToCompressQueryCol(SQInfo *pQInfo);
bool doBuildResCheck(SQInfo* pQInfo);
void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status);
bool onlyQueryTags(STaskAttr* pQueryAttr);
void destroyUdfInfo(struct SUdfInfo* pUdfInfo);
//void destroyUdfInfo(struct SUdfInfo* pUdfInfo);
bool isValidQInfo(void *param);
@ -644,5 +640,7 @@ void freeQueryAttr(STaskAttr *pQuery);
int32_t getMaximumIdleDurationSec();
void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type);
void setTaskStatus(SExecTaskInfo *pTaskInfo, int8_t status);
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle);
#endif // TDENGINE_EXECUTORIMPL_H

8
source/libs/executor/src/executil.c
View File

@ -547,7 +547,7 @@ static UNUSED_FUNC int32_t mergeIntoGroupResultImpl(STaskRuntimeEnv *pRuntimeEnv
pTableQueryInfoList = malloc(POINTER_BYTES * size);
if (pTableQueryInfoList == NULL || posList == NULL || pGroupResInfo->pRows == NULL || pGroupResInfo->pRows == NULL) {
// qError("QInfo:%"PRIu64" failed alloc memory", GET_QID(pRuntimeEnv));
// qError("QInfo:%"PRIu64" failed alloc memory", GET_TASKID(pRuntimeEnv));
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _end;
}
@ -619,7 +619,7 @@ static UNUSED_FUNC int32_t mergeIntoGroupResultImpl(STaskRuntimeEnv *pRuntimeEnv
int64_t endt = taosGetTimestampMs();
// qDebug("QInfo:%"PRIx64" result merge completed for group:%d, elapsed time:%" PRId64 " ms", GET_QID(pRuntimeEnv),
// qDebug("QInfo:%"PRIx64" result merge completed for group:%d, elapsed time:%" PRId64 " ms", GET_TASKID(pRuntimeEnv),
// pGroupResInfo->currentGroup, endt - startt);
_end:
@ -641,13 +641,13 @@ int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, STaskRuntimeEnv* pRun
break;
}
// qDebug("QInfo:%"PRIu64" no result in group %d, continue", GET_QID(pRuntimeEnv), pGroupResInfo->currentGroup);
// qDebug("QInfo:%"PRIu64" no result in group %d, continue", GET_TASKID(pRuntimeEnv), pGroupResInfo->currentGroup);
cleanupGroupResInfo(pGroupResInfo);
incNextGroup(pGroupResInfo);
}
// int64_t elapsedTime = taosGetTimestampUs() - st;
// qDebug("QInfo:%"PRIu64" merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", GET_QID(pRuntimeEnv),
// qDebug("QInfo:%"PRIu64" merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", GET_TASKID(pRuntimeEnv),
// pGroupResInfo->currentGroup, pGroupResInfo->totalGroup, elapsedTime);
return TSDB_CODE_SUCCESS;

239
source/libs/executor/src/executorMain.c
View File

@ -13,11 +13,12 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <dataSinkMgt.h>
#include "exception.h"
#include "os.h"
#include "tcache.h"
#include "tglobal.h"
#include "tmsg.h"
#include "exception.h"
#include "thash.h"
#include "executorimpl.h"
@ -66,152 +67,24 @@ void freeParam(STaskParam *param) {
tfree(param->prevResult);
}
// todo parse json to get the operator tree.
int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo) {
assert(tsdb != NULL && pSubplan != NULL);
int32_t qCreateTask(void* tsdb, int32_t vgId, void* pQueryMsg, qTaskInfo_t* pTaskInfo, uint64_t taskId) {
assert(pQueryMsg != NULL && tsdb != NULL);
int32_t code = TSDB_CODE_SUCCESS;
#if 0
STaskParam param = {0};
code = convertQueryMsg(pQueryMsg, &param);
int32_t code = doCreateExecTaskInfo(pSubplan, (SExecTaskInfo**) pTaskInfo, tsdb);
if (code != TSDB_CODE_SUCCESS) {
goto _over;
goto _error;
}
if (pQueryMsg->numOfTables <= 0) {
qError("Invalid number of tables to query, numOfTables:%d", pQueryMsg->numOfTables);
code = TSDB_CODE_QRY_INVALID_MSG;
goto _over;
}
if (param.pTableIdList == NULL || taosArrayGetSize(param.pTableIdList) == 0) {
qError("qmsg:%p, SQueryTableReq wrong format", pQueryMsg);
code = TSDB_CODE_QRY_INVALID_MSG;
goto _over;
}
SQueriedTableInfo info = { .numOfTags = pQueryMsg->numOfTags, .numOfCols = pQueryMsg->numOfCols, .colList = pQueryMsg->tableCols};
if ((code = createQueryFunc(&info, pQueryMsg->numOfOutput, &param.pExprs, param.pExpr, param.pTagColumnInfo,
pQueryMsg->queryType, pQueryMsg, param.pUdfInfo)) != TSDB_CODE_SUCCESS) {
goto _over;
}
if (param.pSecExpr != NULL) {
if ((code = createIndirectQueryFuncExprFromMsg(pQueryMsg, pQueryMsg->secondStageOutput, &param.pSecExprs, param.pSecExpr, param.pExprs, param.pUdfInfo)) != TSDB_CODE_SUCCESS) {
goto _over;
}
}
if (param.colCond != NULL) {
if ((code = createQueryFilter(param.colCond, pQueryMsg->colCondLen, &param.pFilters)) != TSDB_CODE_SUCCESS) {
goto _over;
}
}
param.pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, param.pGroupColIndex, &code);
if ((param.pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) {
goto _over;
}
bool isSTableQuery = false;
STableGroupInfo tableGroupInfo = {0};
int64_t st = taosGetTimestampUs();
if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_TABLE_QUERY)) {
STableIdInfo *id = taosArrayGet(param.pTableIdList, 0);
qDebug("qmsg:%p query normal table, uid:%"PRId64", tid:%d", pQueryMsg, id->uid, id->tid);
if ((code = tsdbGetOneTableGroup(tsdb, id->uid, pQueryMsg->window.skey, &tableGroupInfo)) != TSDB_CODE_SUCCESS) {
goto _over;
}
} else if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY|TSDB_QUERY_TYPE_STABLE_QUERY)) {
isSTableQuery = true;
// also note there's possibility that only one table in the super table
if (!TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY)) {
STableIdInfo *id = taosArrayGet(param.pTableIdList, 0);
// group by normal column, do not pass the group by condition to tsdb to group table into different group
int32_t numOfGroupByCols = pQueryMsg->numOfGroupCols;
if (pQueryMsg->numOfGroupCols == 1 && !TSDB_COL_IS_TAG(param.pGroupColIndex->flag)) {
numOfGroupByCols = 0;
}
qDebug("qmsg:%p query stable, uid:%"PRIu64", tid:%d", pQueryMsg, id->uid, id->tid);
code = tsdbQuerySTableByTagCond(tsdb, id->uid, pQueryMsg->window.skey, param.tagCond, pQueryMsg->tagCondLen,
pQueryMsg->tagNameRelType, param.tbnameCond, &tableGroupInfo, param.pGroupColIndex, numOfGroupByCols);
if (code != TSDB_CODE_SUCCESS) {
qError("qmsg:%p failed to query stable, reason: %s", pQueryMsg, tstrerror(code));
goto _over;
}
} else {
code = tsdbGetTableGroupFromIdList(tsdb, param.pTableIdList, &tableGroupInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _over;
}
qDebug("qmsg:%p query on %u tables in one group from client", pQueryMsg, tableGroupInfo.numOfTables);
}
int64_t el = taosGetTimestampUs() - st;
qDebug("qmsg:%p tag filter completed, numOfTables:%u, elapsed time:%"PRId64"us", pQueryMsg, tableGroupInfo.numOfTables, el);
} else {
assert(0);
}
code = checkForQueryBuf(tableGroupInfo.numOfTables);
if (code != TSDB_CODE_SUCCESS) { // not enough query buffer, abort
goto _over;
}
assert(pQueryMsg->stableQuery == isSTableQuery);
(*pTaskInfo) = createQInfoImpl(pQueryMsg, param.pGroupbyExpr, param.pExprs, param.pSecExprs, &tableGroupInfo,
param.pTagColumnInfo, param.pFilters, vgId, param.sql, qId, param.pUdfInfo);
param.sql = NULL;
param.pExprs = NULL;
param.pSecExprs = NULL;
param.pGroupbyExpr = NULL;
param.pTagColumnInfo = NULL;
param.pFilters = NULL;
if ((*pTaskInfo) == NULL) {
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _over;
}
param.pUdfInfo = NULL;
code = initQInfo(&pQueryMsg->tsBuf, tsdb, NULL, *pTaskInfo, &param, (char*)pQueryMsg, pQueryMsg->prevResultLen, NULL);
_over:
if (param.pGroupbyExpr != NULL) {
taosArrayDestroy(param.pGroupbyExpr->columnInfo);
}
tfree(param.colCond);
destroyUdfInfo(param.pUdfInfo);
taosArrayDestroy(param.pTableIdList);
param.pTableIdList = NULL;
freeParam(&param);
for (int32_t i = 0; i < pQueryMsg->numOfCols; i++) {
SColumnInfo* column = pQueryMsg->tableCols + i;
freeColumnFilterInfo(column->flist.filterInfo, column->flist.numOfFilters);
}
filterFreeInfo(param.pFilters);
//pTaskInfo already freed in initQInfo, but *pTaskInfo may not pointer to null;
SDataSinkMgtCfg cfg = {.maxDataBlockNum = 1000};
code = dsDataSinkMgtInit(&cfg);
if (code != TSDB_CODE_SUCCESS) {
*pTaskInfo = NULL;
goto _error;
}
#endif
DataSinkHandle pHandle = NULL;
code = dsCreateDataSinker(pSubplan->pDataSink, &pHandle);
_error:
// if failed to add ref for all tables in this query, abort current query
return code;
}
@ -250,7 +123,7 @@ int waitMoment(SQInfo* pQInfo){
while(used_ms < ms) {
taosMsleep(1000);
used_ms += 1000;
if(isQueryKilled(pQInfo)){
if(isTaskKilled(pQInfo)){
printf("test check query is canceled, sleep break.%s\n", pQInfo->sql);
break;
}
@ -261,68 +134,64 @@ int waitMoment(SQInfo* pQInfo){
}
#endif
bool qExecTask(qTaskInfo_t qinfo, uint64_t *qId) {
SQInfo *pQInfo = (SQInfo *)qinfo;
assert(pQInfo && pQInfo->signature == pQInfo);
bool qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes) {
SExecTaskInfo *pTaskInfo = (SExecTaskInfo *) tinfo;
int64_t threadId = taosGetSelfPthreadId();
int64_t curOwner = 0;
if ((curOwner = atomic_val_compare_exchange_64(&pQInfo->owner, 0, threadId)) != 0) {
qError("QInfo:0x%"PRIx64"-%p qhandle is now executed by thread:%p", pQInfo->qId, pQInfo, (void*) curOwner);
pQInfo->code = TSDB_CODE_QRY_IN_EXEC;
if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {
qError("QInfo:0x%"PRIx64"-%p qhandle is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*) curOwner);
pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;
return false;
}
*qId = pQInfo->qId;
if(pQInfo->startExecTs == 0)
pQInfo->startExecTs = taosGetTimestampMs();
if (isQueryKilled(pQInfo)) {
qDebug("QInfo:0x%"PRIx64" it is already killed, abort", pQInfo->qId);
return doBuildResCheck(pQInfo);
if(pTaskInfo->cost.start == 0) {
pTaskInfo->cost.start = taosGetTimestampMs();
}
STaskRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
if (pRuntimeEnv->tableqinfoGroupInfo.numOfTables == 0) {
qDebug("QInfo:0x%"PRIx64" no table exists for query, abort", pQInfo->qId);
// setTaskStatus(pRuntimeEnv, QUERY_COMPLETED);
return doBuildResCheck(pQInfo);
if (isTaskKilled(pTaskInfo)) {
qDebug("QInfo:0x%"PRIx64" it is already killed, abort", GET_TASKID(pTaskInfo));
// return doBuildResCheck(pTaskInfo);
}
// STaskRuntimeEnv* pRuntimeEnv = &pTaskInfo->runtimeEnv;
// if (pTaskInfo->tableqinfoGroupInfo.numOfTables == 0) {
// qDebug("QInfo:0x%"PRIx64" no table exists for query, abort", GET_TASKID(pTaskInfo));
// setTaskStatus(pTaskInfo, TASK_COMPLETED);
// return doBuildResCheck(pTaskInfo);
// }
// error occurs, record the error code and return to client
int32_t ret = setjmp(pQInfo->runtimeEnv.env);
int32_t ret = setjmp(pTaskInfo->env);
if (ret != TSDB_CODE_SUCCESS) {
publishQueryAbortEvent(pQInfo, ret);
pQInfo->code = ret;
qDebug("QInfo:0x%"PRIx64" query abort due to error/cancel occurs, code:%s", pQInfo->qId, tstrerror(pQInfo->code));
return doBuildResCheck(pQInfo);
publishQueryAbortEvent(pTaskInfo, ret);
pTaskInfo->code = ret;
qDebug("QInfo:0x%"PRIx64" query abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));
// return doBuildResCheck(pTaskInfo);
}
qDebug("QInfo:0x%"PRIx64" query task is launched", pQInfo->qId);
qDebug("QInfo:0x%"PRIx64" query task is launched", GET_TASKID(pTaskInfo));
bool newgroup = false;
publishOperatorProfEvent(pRuntimeEnv->proot, QUERY_PROF_BEFORE_OPERATOR_EXEC);
publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_BEFORE_OPERATOR_EXEC);
int64_t st = taosGetTimestampUs();
pRuntimeEnv->outputBuf = pRuntimeEnv->proot->exec(pRuntimeEnv->proot, &newgroup);
pQInfo->summary.elapsedTime += (taosGetTimestampUs() - st);
#ifdef TEST_IMPL
waitMoment(pQInfo);
#endif
publishOperatorProfEvent(pRuntimeEnv->proot, QUERY_PROF_AFTER_OPERATOR_EXEC);
pRuntimeEnv->resultInfo.total += GET_NUM_OF_RESULTS(pRuntimeEnv);
*pRes = pTaskInfo->pRoot->exec(pTaskInfo->pRoot, &newgroup);
// todo put the result into sink node.
if (isQueryKilled(pQInfo)) {
qDebug("QInfo:0x%"PRIx64" query is killed", pQInfo->qId);
} else if (GET_NUM_OF_RESULTS(pRuntimeEnv) == 0) {
qDebug("QInfo:0x%"PRIx64" over, %u tables queried, total %"PRId64" rows returned", pQInfo->qId, pRuntimeEnv->tableqinfoGroupInfo.numOfTables,
pRuntimeEnv->resultInfo.total);
pTaskInfo->cost.elapsedTime += (taosGetTimestampUs() - st);
publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_AFTER_OPERATOR_EXEC);
if (isTaskKilled(pTaskInfo)) {
qDebug("QInfo:0x%"PRIx64" query is killed", GET_TASKID(pTaskInfo));
// } else if (GET_NUM_OF_RESULTS(pRuntimeEnv) == 0) {
// qDebug("QInfo:0x%"PRIx64" over, %u tables queried, total %"PRId64" rows returned", pTaskInfo->qId, pRuntimeEnv->tableqinfoGroupInfo.numOfTables,
// pRuntimeEnv->resultInfo.total);
} else {
qDebug("QInfo:0x%"PRIx64" query paused, %d rows returned, total:%" PRId64 " rows", pQInfo->qId,
GET_NUM_OF_RESULTS(pRuntimeEnv), pRuntimeEnv->resultInfo.total);
// qDebug("QInfo:0x%"PRIx64" query paused, %d rows returned, total:%" PRId64 " rows", pTaskInfo->qId,
// GET_NUM_OF_RESULTS(pRuntimeEnv), pRuntimeEnv->resultInfo.total);
}
return doBuildResCheck(pQInfo);
// return doBuildResCheck(pTaskInfo);
}
int32_t qRetrieveQueryResultInfo(qTaskInfo_t qinfo, bool* buildRes, void* pRspContext) {
@ -398,13 +267,13 @@ int32_t qKillTask(qTaskInfo_t qinfo) {
}
int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
SQInfo *pQInfo = (SQInfo *)qinfo;
SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
if (pTaskInfo == NULL /*|| !isValidQInfo(pTaskInfo)*/) {
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
return isQueryKilled(pQInfo) || Q_STATUS_EQUAL(pQInfo->runtimeEnv.status, QUERY_OVER);
return isTaskKilled(pTaskInfo) || Q_STATUS_EQUAL(pTaskInfo->status, TASK_OVER);
}
void qDestroyTask(qTaskInfo_t qHandle) {

323
source/libs/executor/src/executorimpl.c
View File

@ -26,6 +26,7 @@
#include "thash.h"
#include "ttypes.h"
#include "query.h"
#include "tsdb.h"
#define IS_MAIN_SCAN(runtime) ((runtime)->scanFlag == MAIN_SCAN)
#define IS_REVERSE_SCAN(runtime) ((runtime)->scanFlag == REVERSE_SCAN)
@ -204,12 +205,10 @@ static void destroyStateWindowOperatorInfo(void* param, int32_t numOfOutput);
static void destroyAggOperatorInfo(void* param, int32_t numOfOutput);
static void destroyOperatorInfo(SOperatorInfo* pOperator);
void setTaskStatus(STaskInfo *pTaskInfo, int8_t status);
static void doSetOperatorCompleted(SOperatorInfo* pOperator) {
pOperator->status = OP_EXEC_DONE;
if (pOperator->pTaskInfo != NULL) {
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
}
}
@ -1509,7 +1508,7 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SGroupbyOperatorInfo *pIn
int16_t type = pColInfoData->info.type;
if (type == TSDB_DATA_TYPE_FLOAT || type == TSDB_DATA_TYPE_DOUBLE) {
//qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_QID(pRuntimeEnv));
//qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_TASKID(pRuntimeEnv));
return;
}
@ -1908,7 +1907,7 @@ static SQLFunctionCtx* createSQLFunctionCtx(STaskRuntimeEnv* pRuntimeEnv, SExprI
pCtx->param[2].i = pQueryAttr->window.ekey;
pCtx->param[2].nType = TSDB_DATA_TYPE_BIGINT;
} else if (functionId == FUNCTION_ARITHM) {
pCtx->param[1].pz = (char*) getScalarFuncSupport(pRuntimeEnv->scalarSup, i);
// pCtx->param[1].pz = (char*) getScalarFuncSupport(pRuntimeEnv->scalarSup, i);
}
}
@ -1940,7 +1939,7 @@ static void* destroySQLFunctionCtx(SQLFunctionCtx* pCtx, int32_t numOfOutput) {
}
static int32_t setupQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv, int32_t numOfTables, SArray* pOperator, void* merger) {
//qDebug("QInfo:0x%"PRIx64" setup runtime env", GET_QID(pRuntimeEnv));
//qDebug("QInfo:0x%"PRIx64" setup runtime env", GET_TASKID(pRuntimeEnv));
STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
pRuntimeEnv->prevGroupId = INT32_MIN;
@ -1977,7 +1976,7 @@ static int32_t setupQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv, int32_t numOfT
}
}
//qDebug("QInfo:0x%"PRIx64" init runtime environment completed", GET_QID(pRuntimeEnv));
//qDebug("QInfo:0x%"PRIx64" init runtime environment completed", GET_TASKID(pRuntimeEnv));
// group by normal column, sliding window query, interval query are handled by interval query processor
// interval (down sampling operation)
@ -2164,8 +2163,8 @@ _clean:
static void doFreeQueryHandle(STaskRuntimeEnv* pRuntimeEnv) {
STaskAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
// tsdbCleanupQueryHandle(pRuntimeEnv->pQueryHandle);
pRuntimeEnv->pQueryHandle = NULL;
// tsdbCleanupQueryHandle(pRuntimeEnv->pTsdbReadHandle);
pRuntimeEnv->pTsdbReadHandle = NULL;
// SMemRef* pMemRef = &pQueryAttr->memRef;
// assert(pMemRef->ref == 0 && pMemRef->snapshot.imem == NULL && pMemRef->snapshot.mem == NULL);
@ -2191,7 +2190,7 @@ static void teardownQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv) {
//qDebug("QInfo:0x%"PRIx64" teardown runtime env", pQInfo->qId);
destroyScalarFuncSupport(pRuntimeEnv->scalarSup, pQueryAttr->numOfOutput);
destroyUdfInfo(pRuntimeEnv->pUdfInfo);
// destroyUdfInfo(pRuntimeEnv->pUdfInfo);
destroyResultBuf(pRuntimeEnv->pResultBuf);
doFreeQueryHandle(pRuntimeEnv);
@ -2224,18 +2223,18 @@ static bool needBuildResAfterQueryComplete(SQInfo* pQInfo) {
return pQInfo->rspContext != NULL;
}
bool isQueryKilled(SQInfo *pQInfo) {
if (IS_QUERY_KILLED(pQInfo)) {
bool isTaskKilled(SExecTaskInfo *pTaskInfo) {
if (IS_QUERY_KILLED(pTaskInfo)) {
return true;
}
// query has been executed more than tsShellActivityTimer, and the retrieve has not arrived
// abort current query execution.
if (pQInfo->owner != 0 && ((taosGetTimestampSec() - pQInfo->startExecTs/1000) > getMaximumIdleDurationSec()) &&
(!needBuildResAfterQueryComplete(pQInfo))) {
if (pTaskInfo->owner != 0 && ((taosGetTimestampSec() - pTaskInfo->cost.start/1000) > 10*getMaximumIdleDurationSec())
/*(!needBuildResAfterQueryComplete(pTaskInfo))*/) {
assert(pQInfo->startExecTs != 0);
//qDebug("QInfo:%" PRIu64 " retrieve not arrive beyond %d ms, abort current query execution, start:%" PRId64
assert(pTaskInfo->cost.start != 0);
// qDebug("QInfo:%" PRIu64 " retrieve not arrive beyond %d ms, abort current query execution, start:%" PRId64
// ", current:%d", pQInfo->qId, 1, pQInfo->startExecTs, taosGetTimestampSec());
return true;
}
@ -2887,7 +2886,7 @@ void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFi
}
}
int32_t loadDataBlock(STaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
int32_t loadDataBlock(SExecTaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
STaskCostInfo* pCost = &pTaskInfo->cost;
pCost->totalBlocks += 1;
@ -2896,13 +2895,15 @@ int32_t loadDataBlock(STaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDa
pCost->totalCheckedRows += pBlock->info.rows;
pCost->loadBlocks += 1;
// pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pQueryHandle, NULL);
pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
if (pBlock->pDataBlock == NULL) {
return terrno;
} else {
return TSDB_CODE_SUCCESS;
}
}
int32_t loadDataBlockOnDemand(STaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
int32_t loadDataBlockOnDemand(SExecTaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
*status = BLK_DATA_NO_NEEDED;
pBlock->pDataBlock = NULL;
@ -2991,10 +2992,10 @@ int32_t loadDataBlockOnDemand(STaskInfo *pTaskInfo, STableScanInfo* pTableScanIn
} else if ((*status) == BLK_DATA_STATIS_NEEDED) {
// this function never returns error?
pCost->loadBlockStatis += 1;
// tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->pQueryHandle, &pBlock->pBlockAgg);
// tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
if (pBlock->pBlockAgg == NULL) { // data block statistics does not exist, load data block
// pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pQueryHandle, NULL);
// pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
pCost->totalCheckedRows += pBlock->info.rows;
}
} else {
@ -3002,7 +3003,7 @@ int32_t loadDataBlockOnDemand(STaskInfo *pTaskInfo, STableScanInfo* pTableScanIn
// load the data block statistics to perform further filter
pCost->loadBlockStatis += 1;
// tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->pQueryHandle, &pBlock->pBlockAgg);
// tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
if (pQueryAttr->topBotQuery && pBlock->pBlockAgg != NULL) {
{ // set previous window
@ -3048,7 +3049,7 @@ int32_t loadDataBlockOnDemand(STaskInfo *pTaskInfo, STableScanInfo* pTableScanIn
pCost->totalCheckedRows += pBlockInfo->rows;
pCost->loadBlocks += 1;
// pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pQueryHandle, NULL);
// pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
// if (pBlock->pDataBlock == NULL) {
// return terrno;
// }
@ -3449,12 +3450,12 @@ void initCtxOutputBuffer(SQLFunctionCtx* pCtx, int32_t size) {
}
}
void setTaskStatus(STaskInfo *pTaskInfo, int8_t status) {
if (status == QUERY_NOT_COMPLETED) {
void setTaskStatus(SExecTaskInfo *pTaskInfo, int8_t status) {
if (status == TASK_NOT_COMPLETED) {
pTaskInfo->status = status;
} else {
// QUERY_NOT_COMPLETED is not compatible with any other status, so clear its position first
CLEAR_QUERY_STATUS(pTaskInfo, QUERY_NOT_COMPLETED);
CLEAR_QUERY_STATUS(pTaskInfo, TASK_NOT_COMPLETED);
pTaskInfo->status |= status;
}
}
@ -3476,6 +3477,10 @@ static void setupEnvForReverseScan(STableScanInfo *pTableScanInfo, SQLFunctionCt
SWITCH_ORDER(pTableScanInfo->order);
setupQueryRangeForReverseScan(pTableScanInfo);
pTableScanInfo->times = 1;
pTableScanInfo->current = 0;
pTableScanInfo->reverseTimes = 0;
}
void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset) {
@ -3700,10 +3705,10 @@ void setCtxTagForJoin(STaskRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx, SExprI
//
// int16_t tagType = pCtx[0].tag.nType;
// if (tagType == TSDB_DATA_TYPE_BINARY || tagType == TSDB_DATA_TYPE_NCHAR) {
// //qDebug("QInfo:0x%"PRIx64" set tag value for join comparison, colId:%" PRId64 ", val:%s", GET_QID(pRuntimeEnv),
// //qDebug("QInfo:0x%"PRIx64" set tag value for join comparison, colId:%" PRId64 ", val:%s", GET_TASKID(pRuntimeEnv),
//// pExprInfo->base.param[0].i, pCtx[0].tag.pz);
// } else {
// //qDebug("QInfo:0x%"PRIx64" set tag value for join comparison, colId:%" PRId64 ", val:%" PRId64, GET_QID(pRuntimeEnv),
// //qDebug("QInfo:0x%"PRIx64" set tag value for join comparison, colId:%" PRId64 ", val:%" PRId64, GET_TASKID(pRuntimeEnv),
//// pExprInfo->base.param[0].i, pCtx[0].tag.i);
// }
// }
@ -3723,9 +3728,9 @@ int32_t setTimestampListJoinInfo(STaskRuntimeEnv* pRuntimeEnv, SVariant* pTag, S
// failed to find data with the specified tag value and vnodeId
if (!tsBufIsValidElem(&elem)) {
if (pTag->nType == TSDB_DATA_TYPE_BINARY || pTag->nType == TSDB_DATA_TYPE_NCHAR) {
//qError("QInfo:0x%"PRIx64" failed to find tag:%s in ts_comp", GET_QID(pRuntimeEnv), pTag->pz);
//qError("QInfo:0x%"PRIx64" failed to find tag:%s in ts_comp", GET_TASKID(pRuntimeEnv), pTag->pz);
} else {
//qError("QInfo:0x%"PRIx64" failed to find tag:%" PRId64 " in ts_comp", GET_QID(pRuntimeEnv), pTag->i);
//qError("QInfo:0x%"PRIx64" failed to find tag:%" PRId64 " in ts_comp", GET_TASKID(pRuntimeEnv), pTag->i);
}
return -1;
@ -3734,17 +3739,17 @@ int32_t setTimestampListJoinInfo(STaskRuntimeEnv* pRuntimeEnv, SVariant* pTag, S
// Keep the cursor info of current table
pTableQueryInfo->cur = tsBufGetCursor(pRuntimeEnv->pTsBuf);
if (pTag->nType == TSDB_DATA_TYPE_BINARY || pTag->nType == TSDB_DATA_TYPE_NCHAR) {
//qDebug("QInfo:0x%"PRIx64" find tag:%s start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_QID(pRuntimeEnv), pTag->pz, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
//qDebug("QInfo:0x%"PRIx64" find tag:%s start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_TASKID(pRuntimeEnv), pTag->pz, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
} else {
//qDebug("QInfo:0x%"PRIx64" find tag:%"PRId64" start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_QID(pRuntimeEnv), pTag->i, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
//qDebug("QInfo:0x%"PRIx64" find tag:%"PRId64" start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_TASKID(pRuntimeEnv), pTag->i, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
}
} else {
tsBufSetCursor(pRuntimeEnv->pTsBuf, &pTableQueryInfo->cur);
if (pTag->nType == TSDB_DATA_TYPE_BINARY || pTag->nType == TSDB_DATA_TYPE_NCHAR) {
//qDebug("QInfo:0x%"PRIx64" find tag:%s start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_QID(pRuntimeEnv), pTag->pz, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
//qDebug("QInfo:0x%"PRIx64" find tag:%s start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_TASKID(pRuntimeEnv), pTag->pz, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
} else {
//qDebug("QInfo:0x%"PRIx64" find tag:%"PRId64" start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_QID(pRuntimeEnv), pTag->i, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
//qDebug("QInfo:0x%"PRIx64" find tag:%"PRId64" start pos in ts_comp, blockIndex:%d, tsIndex:%d", GET_TASKID(pRuntimeEnv), pTag->i, pTableQueryInfo->cur.blockIndex, pTableQueryInfo->cur.tsIndex);
}
}
@ -3882,7 +3887,7 @@ static int32_t doCopyToSDataBlock(STaskRuntimeEnv* pRuntimeEnv, SGroupResInfo* p
int32_t start = 0;
int32_t step = -1;
//qDebug("QInfo:0x%"PRIx64" start to copy data from windowResInfo to output buf", GET_QID(pRuntimeEnv));
//qDebug("QInfo:0x%"PRIx64" start to copy data from windowResInfo to output buf", GET_TASKID(pRuntimeEnv));
assert(orderType == TSDB_ORDER_ASC || orderType == TSDB_ORDER_DESC);
if (orderType == TSDB_ORDER_ASC) {
@ -3927,7 +3932,7 @@ static int32_t doCopyToSDataBlock(STaskRuntimeEnv* pRuntimeEnv, SGroupResInfo* p
}
}
//qDebug("QInfo:0x%"PRIx64" copy data to query buf completed", GET_QID(pRuntimeEnv));
//qDebug("QInfo:0x%"PRIx64" copy data to query buf completed", GET_TASKID(pRuntimeEnv));
pBlock->info.rows = numOfResult;
return 0;
}
@ -4055,7 +4060,7 @@ static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data
//qDebug("QInfo:0x%"PRIx64" set %d subscribe info", pQInfo->qId, total);
// Check if query is completed or not for stable query or normal table query respectively.
if (Q_STATUS_EQUAL(pRuntimeEnv->status, QUERY_COMPLETED) && pRuntimeEnv->proot->status == OP_EXEC_DONE) {
if (Q_STATUS_EQUAL(pRuntimeEnv->status, TASK_COMPLETED) && pRuntimeEnv->proot->status == OP_EXEC_DONE) {
// setTaskStatus(pOperator->pTaskInfo, QUERY_OVER);
}
}
@ -4232,7 +4237,7 @@ void queryCostStatis(SQInfo *pQInfo) {
//
// assert(pQueryAttr->pos >= 0 && pQueryAttr->pos <= pBlockInfo->rows - 1);
//
// SArray * pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL);
// SArray * pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
// SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
//
// // update the pQueryAttr->limit.offset value, and pQueryAttr->pos value
@ -4244,7 +4249,7 @@ void queryCostStatis(SQInfo *pQInfo) {
//
// int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
//
// //qDebug("QInfo:0x%"PRIx64" check data block, brange:%" PRId64 "-%" PRId64 ", numBlocksOfStep:%d, numOfRes:%d, lastKey:%"PRId64, GET_QID(pRuntimeEnv),
// //qDebug("QInfo:0x%"PRIx64" check data block, brange:%" PRId64 "-%" PRId64 ", numBlocksOfStep:%d, numOfRes:%d, lastKey:%"PRId64, GET_TASKID(pRuntimeEnv),
// pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes, pQuery->current->lastKey);
//}
@ -4259,22 +4264,22 @@ void queryCostStatis(SQInfo *pQInfo) {
// int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQueryAttr->order.order);
//
// STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
// TsdbQueryHandleT pQueryHandle = pRuntimeEnv->pQueryHandle;
// TsdbQueryHandleT pTsdbReadHandle = pRuntimeEnv->pTsdbReadHandle;
//
// SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
// while (tsdbNextDataBlock(pQueryHandle)) {
// if (isQueryKilled(pRuntimeEnv->qinfo)) {
// while (tsdbNextDataBlock(pTsdbReadHandle)) {
// if (isTaskKilled(pRuntimeEnv->qinfo)) {
// longjmp(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
// }
//
// tsdbRetrieveDataBlockInfo(pQueryHandle, &blockInfo);
// tsdbRetrieveDataBlockInfo(pTsdbReadHandle, &blockInfo);
//
// if (pQueryAttr->limit.offset > blockInfo.rows) {
// pQueryAttr->limit.offset -= blockInfo.rows;
// pTableQueryInfo->lastKey = (QUERY_IS_ASC_QUERY(pQueryAttr)) ? blockInfo.window.ekey : blockInfo.window.skey;
// pTableQueryInfo->lastKey += step;
//
// //qDebug("QInfo:0x%"PRIx64" skip rows:%d, offset:%" PRId64, GET_QID(pRuntimeEnv), blockInfo.rows,
// //qDebug("QInfo:0x%"PRIx64" skip rows:%d, offset:%" PRId64, GET_TASKID(pRuntimeEnv), blockInfo.rows,
// pQuery->limit.offset);
// } else { // find the appropriated start position in current block
// updateOffsetVal(pRuntimeEnv, &blockInfo);
@ -4300,7 +4305,7 @@ void queryCostStatis(SQInfo *pQInfo) {
//
// // load the data block and check data remaining in current data block
// // TODO optimize performance
// SArray * pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL);
// SArray * pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
// SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
//
// tw = *win;
@ -4323,7 +4328,7 @@ void queryCostStatis(SQInfo *pQInfo) {
// pRuntimeEnv->resultRowInfo.curIndex = index; // restore the window index
//
// //qDebug("QInfo:0x%"PRIx64" check data block, brange:%" PRId64 "-%" PRId64 ", numOfRows:%d, numOfRes:%d, lastKey:%" PRId64,
// GET_QID(pRuntimeEnv), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes,
// GET_TASKID(pRuntimeEnv), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes,
// pQueryAttr->current->lastKey);
//
// return key;
@ -4365,8 +4370,8 @@ void queryCostStatis(SQInfo *pQInfo) {
// STableQueryInfo *pTableQueryInfo = pRuntimeEnv->current;
//
// SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
// while (tsdbNextDataBlock(pRuntimeEnv->pQueryHandle)) {
// tsdbRetrieveDataBlockInfo(pRuntimeEnv->pQueryHandle, &blockInfo);
// while (tsdbNextDataBlock(pRuntimeEnv->pTsdbReadHandle)) {
// tsdbRetrieveDataBlockInfo(pRuntimeEnv->pTsdbReadHandle, &blockInfo);
//
// if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
// if (pWindowResInfo->prevSKey == TSKEY_INITIAL_VAL) {
@ -4412,7 +4417,7 @@ void queryCostStatis(SQInfo *pQInfo) {
// */
// if ((tw.skey <= blockInfo.window.ekey && ascQuery) || (tw.ekey >= blockInfo.window.skey && !ascQuery)) {
//
// SArray *pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL);
// SArray *pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
// SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
//
// if ((win.ekey > blockInfo.window.ekey && ascQuery) || (win.ekey < blockInfo.window.skey && !ascQuery)) {
@ -4486,7 +4491,7 @@ static int32_t setupQueryHandle(void* tsdb, STaskRuntimeEnv* pRuntimeEnv, int64_
terrno = TSDB_CODE_SUCCESS;
if (isFirstLastRowQuery(pQueryAttr)) {
pRuntimeEnv->pQueryHandle = tsdbQueryLastRow(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
pRuntimeEnv->pTsdbReadHandle = tsdbQueryLastRow(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
// update the query time window
pQueryAttr->window = cond.twindow;
@ -4507,11 +4512,11 @@ static int32_t setupQueryHandle(void* tsdb, STaskRuntimeEnv* pRuntimeEnv, int64_
}
}
} else if (isCachedLastQuery(pQueryAttr)) {
pRuntimeEnv->pQueryHandle = tsdbQueryCacheLast(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
pRuntimeEnv->pTsdbReadHandle = tsdbQueryCacheLast(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
} else if (pQueryAttr->pointInterpQuery) {
pRuntimeEnv->pQueryHandle = tsdbQueryRowsInExternalWindow(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
pRuntimeEnv->pTsdbReadHandle = tsdbQueryRowsInExternalWindow(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
} else {
pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
pRuntimeEnv->pTsdbReadHandle = tsdbQueryTables(tsdb, &cond, &pQueryAttr->tableGroupInfo, qId, &pQueryAttr->memRef);
}
#endif
return terrno;
@ -4543,19 +4548,19 @@ int32_t doInitQInfo(SQInfo* pQInfo, STSBuf* pTsBuf, void* tsdb, void* sourceOptr
switch(tbScanner) {
// case OP_TableBlockInfoScan: {
// pRuntimeEnv->proot = createTableBlockInfoScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv);
// pRuntimeEnv->proot = createTableBlockInfoScanOperator(pRuntimeEnv->pTsdbReadHandle, pRuntimeEnv);
// break;
// }
// case OP_TableSeqScan: {
// pRuntimeEnv->proot = createTableSeqScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv);
// pRuntimeEnv->proot = createTableSeqScanOperator(pRuntimeEnv->pTsdbReadHandle, pRuntimeEnv);
// break;
// }
// case OP_DataBlocksOptScan: {
// pRuntimeEnv->proot = createDataBlocksOptScanInfo(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr), pQueryAttr->needReverseScan? 1:0);
// pRuntimeEnv->proot = createDataBlocksOptScanInfo(pRuntimeEnv->pTsdbReadHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr), pQueryAttr->needReverseScan? 1:0);
// break;
// }
// case OP_TableScan: {
// pRuntimeEnv->proot = createTableScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr));
// pRuntimeEnv->proot = createTableScanOperator(pRuntimeEnv->pTsdbReadHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr));
// break;
// }
default: { // do nothing
@ -4606,7 +4611,7 @@ int32_t doInitQInfo(SQInfo* pQInfo, STSBuf* pTsBuf, void* tsdb, void* sourceOptr
return TSDB_CODE_SUCCESS;
}
static void doTableQueryInfoTimeWindowCheck(STaskInfo* pTaskInfo, STableQueryInfo* pTableQueryInfo, int32_t order) {
static void doTableQueryInfoTimeWindowCheck(SExecTaskInfo* pTaskInfo, STableQueryInfo* pTableQueryInfo, int32_t order) {
if (order == TSDB_ORDER_ASC) {
assert(
(pTableQueryInfo->win.skey <= pTableQueryInfo->win.ekey) &&
@ -4679,20 +4684,20 @@ static SSDataBlock* doTableScanImpl(void* param, bool* newgroup) {
SOperatorInfo *pOperator = (SOperatorInfo*) param;
STableScanInfo *pTableScanInfo = pOperator->info;
STaskInfo *pTaskInfo = pOperator->pTaskInfo;
SExecTaskInfo *pTaskInfo = pOperator->pTaskInfo;
SSDataBlock *pBlock = &pTableScanInfo->block;
STableGroupInfo *pTableGroupInfo = &pOperator->pTaskInfo->tableqinfoGroupInfo;
*newgroup = false;
while (/*tsdbNextDataBlock(pTableScanInfo->pQueryHandle)*/1) {
if (isQueryKilled(pOperator->pRuntimeEnv->qinfo)) {
longjmp(pOperator->pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
while (tsdbNextDataBlock(pTableScanInfo->pTsdbReadHandle)) {
if (isTaskKilled(pOperator->pTaskInfo)) {
longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
pTableScanInfo->numOfBlocks += 1;
// tsdbRetrieveDataBlockInfo(pTableScanInfo->pQueryHandle, &pBlock->info);
tsdbRetrieveDataBlockInfo(pTableScanInfo->pTsdbReadHandle, &pBlock->info);
// todo opt
// if (pTableGroupInfo->numOfTables > 1 || (pRuntimeEnv->current == NULL && pTableGroupInfo->numOfTables == 1)) {
@ -4711,7 +4716,7 @@ static SSDataBlock* doTableScanImpl(void* param, bool* newgroup) {
int32_t code = loadDataBlock(pTaskInfo, pTableScanInfo, pBlock, &status);
// int32_t code = loadDataBlockOnDemand(pOperator->pRuntimeEnv, pTableScanInfo, pBlock, &status);
if (code != TSDB_CODE_SUCCESS) {
longjmp(pOperator->pRuntimeEnv->env, code);
longjmp(pOperator->pTaskInfo->env, code);
}
// current block is ignored according to filter result by block statistics data, continue load the next block
@ -4729,7 +4734,7 @@ static SSDataBlock* doTableScan(void* param, bool *newgroup) {
SOperatorInfo* pOperator = (SOperatorInfo*) param;
STableScanInfo *pTableScanInfo = pOperator->info;
STaskInfo *pTaskInfo = pOperator->pTaskInfo;
SExecTaskInfo *pTaskInfo = pOperator->pTaskInfo;
SResultRowInfo* pResultRowInfo = pTableScanInfo->pResultRowInfo;
*newgroup = false;
@ -4741,7 +4746,7 @@ static SSDataBlock* doTableScan(void* param, bool *newgroup) {
}
if (++pTableScanInfo->current >= pTableScanInfo->times) {
if (pTableScanInfo->reverseTimes <= 0/* || isTsdbCacheLastRow(pTableScanInfo->pQueryHandle)*/) {
if (pTableScanInfo->reverseTimes <= 0/* || isTsdbCacheLastRow(pTableScanInfo->pTsdbReadHandle)*/) {
return NULL;
} else {
break;
@ -4750,9 +4755,9 @@ static SSDataBlock* doTableScan(void* param, bool *newgroup) {
// do prepare for the next round table scan operation
// STsdbQueryCond cond = createTsdbQueryCond(pQueryAttr, &pQueryAttr->window);
// tsdbResetQueryHandle(pTableScanInfo->pQueryHandle, &cond);
// tsdbResetQueryHandle(pTableScanInfo->pTsdbReadHandle, &cond);
setTaskStatus(pTaskInfo, QUERY_NOT_COMPLETED);
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN;
// if (pTaskInfo->pTsBuf) {
@ -4764,8 +4769,8 @@ static SSDataBlock* doTableScan(void* param, bool *newgroup) {
pResultRowInfo->curPos = 0;
}
//qDebug("QInfo:0x%"PRIx64" start to repeat scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
// GET_QID(pRuntimeEnv), cond.twindow.skey, cond.twindow.ekey);
qDebug("QInfo:0x%"PRIx64" start to repeat scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
}
SSDataBlock *p = NULL;
@ -4773,15 +4778,10 @@ static SSDataBlock* doTableScan(void* param, bool *newgroup) {
if (pTableScanInfo->reverseTimes > 0) {
setupEnvForReverseScan(pTableScanInfo, pTableScanInfo->pCtx, pTableScanInfo->numOfOutput);
// STsdbQueryCond cond = createTsdbQueryCond(pQueryAttr, &pQueryAttr->window);
// tsdbResetQueryHandle(pTableScanInfo->pQueryHandle, &cond);
// tsdbResetQueryHandle(pTableScanInfo->pTsdbReadHandle, &cond);
//qDebug("QInfo:0x%"PRIx64" start to reverse scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
// GET_QID(pRuntimeEnv), cond.twindow.skey, cond.twindow.ekey);
pTableScanInfo->times = 1;
pTableScanInfo->current = 0;
pTableScanInfo->reverseTimes = 0;
// pTableScanInfo->order = cond.order;
qDebug("QInfo:0x%"PRIx64" start to reverse scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
if (pResultRowInfo->size > 0) {
pResultRowInfo->curPos = pResultRowInfo->size - 1;
@ -4814,8 +4814,8 @@ static SSDataBlock* doBlockInfoScan(void* param, bool* newgroup) {
tableBlockDist.maxRows = INT_MIN;
tableBlockDist.minRows = INT_MAX;
tsdbGetFileBlocksDistInfo(pTableScanInfo->pQueryHandle, &tableBlockDist);
tableBlockDist.numOfRowsInMemTable = (int32_t) tsdbGetNumOfRowsInMemTable(pTableScanInfo->pQueryHandle);
tsdbGetFileBlocksDistInfo(pTableScanInfo->pTsdbReadHandle, &tableBlockDist);
tableBlockDist.numOfRowsInMemTable = (int32_t) tsdbGetNumOfRowsInMemTable(pTableScanInfo->pTsdbReadHandle);
SSDataBlock* pBlock = &pTableScanInfo->block;
pBlock->info.rows = 1;
@ -4842,11 +4842,11 @@ static SSDataBlock* doBlockInfoScan(void* param, bool* newgroup) {
}
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime) {
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo) {
assert(repeatTime > 0 && numOfOutput > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = 0;
pInfo->order = order;
@ -4862,6 +4862,7 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, in
pOperator->numOfOutput = numOfOutput;
pOperator->pRuntimeEnv = NULL;
pOperator->exec = doTableScan;
pOperator->pTaskInfo = pTaskInfo;
return pOperator;
}
@ -4869,7 +4870,7 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, in
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv) {
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = 1;
pInfo->reverseTimes = 0;
pInfo->order = pRuntimeEnv->pQueryAttr->order.order;
@ -4893,7 +4894,7 @@ SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, STaskRuntimeEn
SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv) {
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
SColumnInfoData infoData = {{0}};
@ -4977,7 +4978,7 @@ SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, STaskRuntimeE
assert(repeatTime > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = reverseTime;
pInfo->current = 0;
@ -5145,7 +5146,7 @@ SOperatorInfo* createGlobalAggregateOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, S
pOperator->numOfOutput = numOfOutput;
pOperator->pRuntimeEnv = pRuntimeEnv;
pOperator->exec = doGlobalAggregate;
// pOperator->exec = doGlobalAggregate;
pOperator->cleanup = destroyGlobalAggOperatorInfo;
appendUpstream(pOperator, downstream);
@ -5188,7 +5189,7 @@ SOperatorInfo *createMultiwaySortOperatorInfo(STaskRuntimeEnv *pRuntimeEnv, SExp
pOperator->pRuntimeEnv = pRuntimeEnv;
pOperator->numOfOutput = numOfOutput;
pOperator->pExpr = pExpr;
pOperator->exec = doMultiwayMergeSort;
// pOperator->exec = doMultiwayMergeSort;
pOperator->cleanup = destroyGlobalAggOperatorInfo;
return pOperator;
}
@ -5476,7 +5477,7 @@ static SSDataBlock* doProjectOperation(void* param, bool* newgroup) {
assert(*newgroup == false);
*newgroup = prevVal;
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
break;
}
@ -5644,7 +5645,7 @@ static SSDataBlock* doIntervalAgg(void* param, bool* newgroup) {
pOperator->status = OP_RES_TO_RETURN;
closeAllResultRows(&pIntervalInfo->resultRowInfo);
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
finalizeQueryResult(pOperator, pIntervalInfo->pCtx, &pIntervalInfo->resultRowInfo, pIntervalInfo->rowCellInfoOffset);
initGroupResInfo(&pRuntimeEnv->groupResInfo, &pIntervalInfo->resultRowInfo);
@ -5704,7 +5705,7 @@ static SSDataBlock* doAllIntervalAgg(void* param, bool* newgroup) {
pOperator->status = OP_RES_TO_RETURN;
closeAllResultRows(&pIntervalInfo->resultRowInfo);
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
finalizeQueryResult(pOperator, pIntervalInfo->pCtx, &pIntervalInfo->resultRowInfo, pIntervalInfo->rowCellInfoOffset);
initGroupResInfo(&pRuntimeEnv->groupResInfo, &pIntervalInfo->resultRowInfo);
@ -5767,7 +5768,7 @@ static SSDataBlock* doSTableIntervalAgg(void* param, bool* newgroup) {
pOperator->status = OP_RES_TO_RETURN;
pQueryAttr->order.order = order; // TODO : restore the order
doCloseAllTimeWindow(pRuntimeEnv);
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
copyToSDataBlock(pRuntimeEnv, 3000, pIntervalInfo->pRes, pIntervalInfo->rowCellInfoOffset);
if (pIntervalInfo->pRes->info.rows == 0 || !hasRemainData(&pRuntimeEnv->groupResInfo)) {
@ -5822,7 +5823,7 @@ static SSDataBlock* doAllSTableIntervalAgg(void* param, bool* newgroup) {
pOperator->status = OP_RES_TO_RETURN;
pQueryAttr->order.order = order; // TODO : restore the order
doCloseAllTimeWindow(pRuntimeEnv);
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
int64_t st = taosGetTimestampUs();
copyToSDataBlock(pRuntimeEnv, 3000, pIntervalInfo->pRes, pIntervalInfo->rowCellInfoOffset);
@ -5956,7 +5957,7 @@ static SSDataBlock* doStateWindowAgg(void *param, bool* newgroup) {
pOperator->status = OP_RES_TO_RETURN;
closeAllResultRows(&pBInfo->resultRowInfo);
setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
finalizeQueryResult(pOperator, pBInfo->pCtx, &pBInfo->resultRowInfo, pBInfo->rowCellInfoOffset);
initGroupResInfo(&pRuntimeEnv->groupResInfo, &pBInfo->resultRowInfo);
@ -6094,7 +6095,7 @@ static SSDataBlock* hashGroupbyAggregate(void* param, bool* newgroup) {
static void doHandleRemainBlockForNewGroupImpl(SFillOperatorInfo *pInfo, STaskRuntimeEnv* pRuntimeEnv, bool* newgroup) {
pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
int64_t ekey = Q_STATUS_EQUAL(pRuntimeEnv->status, QUERY_COMPLETED)?pRuntimeEnv->pQueryAttr->window.ekey:pInfo->existNewGroupBlock->info.window.ekey;
int64_t ekey = Q_STATUS_EQUAL(pRuntimeEnv->status, TASK_COMPLETED)?pRuntimeEnv->pQueryAttr->window.ekey:pInfo->existNewGroupBlock->info.window.ekey;
taosResetFillInfo(pInfo->pFillInfo, getFillInfoStart(pInfo->pFillInfo));
taosFillSetStartInfo(pInfo->pFillInfo, pInfo->existNewGroupBlock->info.rows, ekey);
@ -6422,7 +6423,7 @@ SOperatorInfo* createFilterOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
SFilterOperatorInfo* pInfo = calloc(1, sizeof(SFilterOperatorInfo));
assert(numOfFilter > 0 && pCols != NULL);
doCreateFilterInfo(pCols, numOfOutput, numOfFilter, &pInfo->pFilterInfo, 0);
// doCreateFilterInfo(pCols, numOfOutput, numOfFilter, &pInfo->pFilterInfo, 0);
pInfo->numOfFilterCols = numOfFilter;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
@ -6714,10 +6715,10 @@ SOperatorInfo* createSLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
pOperator->name = "SLimitOperator";
// pOperator->operatorType = OP_SLimit;
pOperator->operatorType = OP_SLimit;
pOperator->blockingOptr = false;
pOperator->status = OP_IN_EXECUTING;
pOperator->exec = doSLimit;
// pOperator->exec = doSLimit;
pOperator->info = pInfo;
pOperator->pRuntimeEnv = pRuntimeEnv;
pOperator->cleanup = destroySlimitOperatorInfo;
@ -6799,14 +6800,14 @@ static SSDataBlock* doTagScan(void* param, bool* newgroup) {
count += 1;
}
//qDebug("QInfo:0x%"PRIx64" create (tableId, tag) info completed, rows:%d", GET_QID(pRuntimeEnv), count);
//qDebug("QInfo:0x%"PRIx64" create (tableId, tag) info completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
} else if (functionId == FUNCTION_COUNT) {// handle the "count(tbname)" query
SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
*(int64_t*)pColInfo->pData = pInfo->totalTables;
count = 1;
pOperator->status = OP_EXEC_DONE;
//qDebug("QInfo:0x%"PRIx64" create count(tbname) query, res:%d rows:1", GET_QID(pRuntimeEnv), count);
//qDebug("QInfo:0x%"PRIx64" create count(tbname) query, res:%d rows:1", GET_TASKID(pRuntimeEnv), count);
} else { // return only the tags|table name etc.
SExprInfo* pExprInfo = &pOperator->pExpr[0]; // todo use the column list instead of exprinfo
@ -6845,11 +6846,11 @@ static SSDataBlock* doTagScan(void* param, bool* newgroup) {
pOperator->status = OP_EXEC_DONE;
}
//qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_QID(pRuntimeEnv), count);
//qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
}
if (pOperator->status == OP_EXEC_DONE) {
setTaskStatus(pOperator->pRuntimeEnv, QUERY_COMPLETED);
setTaskStatus(pOperator->pRuntimeEnv, TASK_COMPLETED);
}
pRes->info.rows = count;
@ -7174,90 +7175,52 @@ static int32_t deserializeColFilterInfo(SColumnFilterInfo* pColFilters, int16_t
return TSDB_CODE_SUCCESS;
}
/**
* {
"Id": {
"QueryId": 20,
"TemplateId": 0,
"SubplanId": 0
},
"Node": {
"Name": "TableScan",
"InputSchema": [{
"Type": 9,
"ColId": 1,
"Bytes": 8
}, {
"Type": 4,
"ColId": 2,
"Bytes": 4
}, {
"Type": 8,
"ColId": 3,
"Bytes": 20
}],
"TableScan": {
"TableId": 1,
"TableType": 3,
"Flag": 0,
"Window": {
"StartKey": 0,
"EndKey": 0
}
}
},
"DataSink": {
"Name": "Dispatch",
"Dispatch": {
}
}
static SExecTaskInfo* createExecTaskInfo(uint64_t queryId) {
SExecTaskInfo* pTaskInfo = calloc(1, sizeof(SExecTaskInfo));
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pthread_mutex_init(&pTaskInfo->lock, NULL);
pTaskInfo->cost.created = taosGetTimestampMs();
return pTaskInfo;
}
*/
int32_t parseTaskInfo(const char* msg, int32_t len) {
cJSON* pJson = cJSON_Parse(msg);
if (NULL == pJson) {
return TSDB_CODE_INVALID_MSG;
SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTaskInfo, void* param) {
if (pPhyNode->pChildren == NULL || taosArrayGetSize(pPhyNode->pChildren) == 0) {
if (pPhyNode->info.type == OP_TableScan) {
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
SOperatorInfo* pOperatorInfo = createTableScanOperator(param, TSDB_ORDER_ASC, numOfCols, 1, pTaskInfo);
pTaskInfo->pRoot = pOperatorInfo;
}
}
}
cJSON* pSub = cJSON_GetObjectItem(pJson, "ID");
if (NULL != pSub) {
printf("Id : %s\n", pSub->valuestring);
}
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle) {
STsdbQueryCond cond = {.order = TSDB_ORDER_ASC, .numOfCols = 2, .loadExternalRows = false};
cond.twindow.skey = INT64_MIN;
cond.twindow.ekey = INT64_MAX;
cond.colList = calloc(cond.numOfCols, sizeof(SColumnInfo));
cJSON* pNode = cJSON_GetObjectItem(pJson, "Node");
if (pNode == NULL) {
return TSDB_CODE_INVALID_MSG;
}
// todo set the correct table column info
cond.colList[0].type = TSDB_DATA_TYPE_TIMESTAMP;
cond.colList[0].bytes = sizeof(uint64_t);
cond.colList[0].colId = 1;
cJSON* pNodeName = cJSON_GetObjectItem(pNode, "name");
if (pNodeName == NULL) {
return TSDB_CODE_INVALID_MSG;
}
cond.colList[1].type = TSDB_DATA_TYPE_INT;
cond.colList[1].bytes = sizeof(int32_t);
cond.colList[1].colId = 2;
printf("node name is: %s\n", pNodeName->valuestring);
cJSON* pNodeSchema = cJSON_GetObjectItem(pNode, "InputSchema");
if (pNodeSchema == NULL) {
return TSDB_CODE_INVALID_MSG;
}
cJSON* pOperator = cJSON_GetObjectItem(pNode, pNodeName->valuestring);
if (pOperator == NULL) {
return TSDB_CODE_INVALID_MSG;
}
cJSON* pTableId = cJSON_GetObjectItem(pOperator, "tableId");
if (pTableId == NULL) {
return TSDB_CODE_INVALID_MSG;
}
cJSON* pTimeWindow = cJSON_GetObjectItem(pOperator, "window");
if (pTimeWindow == NULL) {
return TSDB_CODE_INVALID_MSG;
}
STableGroupInfo group = {.numOfTables = 1, .pGroupList = taosArrayInit(1, POINTER_BYTES)};
SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo));
STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = 1};
taosArrayPush(pa, &info);
taosArrayPush(group.pGroupList, &pa);
*pTaskInfo = createExecTaskInfo((uint64_t)pPlan->id.queryId);
tsdbReadHandleT tsdbReadHandle = tsdbQueryTables(readerHandle, &cond, &group, (*pTaskInfo)->id.queryId, NULL);
doCreateOperatorTreeNode(pPlan->pNode, *pTaskInfo, tsdbReadHandle);
return TSDB_CODE_SUCCESS;
}
/**
@ -7704,7 +7667,7 @@ int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExp
*pExprInfo = NULL;
int32_t code = TSDB_CODE_SUCCESS;
code = initUdfInfo(pUdfInfo);
// code = initUdfInfo(pUdfInfo);
if (code) {
return code;
}
@ -7996,8 +7959,8 @@ int32_t createFilterInfo(STaskAttr* pQueryAttr, uint64_t qId) {
return TSDB_CODE_SUCCESS;
}
doCreateFilterInfo(pQueryAttr->tableCols, pQueryAttr->numOfCols, pQueryAttr->numOfFilterCols,
&pQueryAttr->pFilterInfo, qId);
// doCreateFilterInfo(pQueryAttr->tableCols, pQueryAttr->numOfCols, pQueryAttr->numOfFilterCols,
// &pQueryAttr->pFilterInfo, qId);
pQueryAttr->createFilterOperator = true;
@ -8505,7 +8468,7 @@ int32_t doDumpQueryResult(SQInfo *pQInfo, char *data, int8_t compressed, int32_t
}
// all data returned, set query over
if (Q_STATUS_EQUAL(pRuntimeEnv->status, QUERY_COMPLETED)) {
if (Q_STATUS_EQUAL(pRuntimeEnv->status, TASK_COMPLETED)) {
// setTaskStatus(pOperator->pTaskInfo, QUERY_OVER);
}
} else {

18
source/libs/executor/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,18 @@
MESSAGE(STATUS "build parser unit test")
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(executorTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(
executorTest
PUBLIC os util common transport gtest taos qcom executor function planner
)
TARGET_INCLUDE_DIRECTORIES(
executorTest
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/executor/"
PRIVATE "${CMAKE_SOURCE_DIR}/source/libs/executor/inc"
)

221
source/libs/executor/test/executorTests.cpp
View File

@ -0,0 +1,221 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <executorimpl.h>
#include <gtest/gtest.h>
#include <tglobal.h>
#include <iostream>
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
#pragma GCC diagnostic ignored "-Wunused-variable"
#pragma GCC diagnostic ignored "-Wsign-compare"
#include "os.h"
#include "taos.h"
#include "tdef.h"
#include "tvariant.h"
#include "tep.h"
#include "trpc.h"
#include "stub.h"
#include "executor.h"
/**
{
"Id": {
"QueryId": 1.3108161807422521e+19,
"TemplateId": 0,
"SubplanId": 0
},
"Node": {
"Name": "TableScan",
"Targets": [{
"Base": {
"Schema": {
"Type": 9,
"ColId": 5000,
"Bytes": 8
},
"Columns": [{
"TableId": 1,
"Flag": 0,
"Info": {
"ColId": 1,
"Type": 9,
"Bytes": 8
}
}],
"InterBytes": 0
},
"Expr": {
"Type": 4,
"Column": {
"Type": 9,
"ColId": 1,
"Bytes": 8
}
}
}, {
"Base": {
"Schema": {
"Type": 4,
"ColId": 5001,
"Bytes": 4
},
"Columns": [{
"TableId": 1,
"Flag": 0,
"Info": {
"ColId": 2,
"Type": 4,
"Bytes": 4
}
}],
"InterBytes": 0
},
"Expr": {
"Type": 4,
"Column": {
"Type": 4,
"ColId": 2,
"Bytes": 4
}
}
}],
"InputSchema": [{
"Type": 9,
"ColId": 5000,
"Bytes": 8
}, {
"Type": 4,
"ColId": 5001,
"Bytes": 4
}],
"TableScan": {
"TableId": 1,
"TableType": 2,
"Flag": 0,
"Window": {
"StartKey": -9.2233720368547758e+18,
"EndKey": 9.2233720368547758e+18
}
}
},
"DataSink": {
"Name": "Dispatch",
"Dispatch": {
}
}
}
*/
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
TEST(testCase, build_executor_tree_Test) {
const char* msg = "{\n"
"\t\"Id\":\t{\n"
"\t\t\"QueryId\":\t1.3108161807422521e+19,\n"
"\t\t\"TemplateId\":\t0,\n"
"\t\t\"SubplanId\":\t0\n"
"\t},\n"
"\t\"Node\":\t{\n"
"\t\t\"Name\":\t\"TableScan\",\n"
"\t\t\"Targets\":\t[{\n"
"\t\t\t\t\"Base\":\t{\n"
"\t\t\t\t\t\"Schema\":\t{\n"
"\t\t\t\t\t\t\"Type\":\t9,\n"
"\t\t\t\t\t\t\"ColId\":\t5000,\n"
"\t\t\t\t\t\t\"Bytes\":\t8\n"
"\t\t\t\t\t},\n"
"\t\t\t\t\t\"Columns\":\t[{\n"
"\t\t\t\t\t\t\t\"TableId\":\t1,\n"
"\t\t\t\t\t\t\t\"Flag\":\t0,\n"
"\t\t\t\t\t\t\t\"Info\":\t{\n"
"\t\t\t\t\t\t\t\t\"ColId\":\t1,\n"
"\t\t\t\t\t\t\t\t\"Type\":\t9,\n"
"\t\t\t\t\t\t\t\t\"Bytes\":\t8\n"
"\t\t\t\t\t\t\t}\n"
"\t\t\t\t\t\t}],\n"
"\t\t\t\t\t\"InterBytes\":\t0\n"
"\t\t\t\t},\n"
"\t\t\t\t\"Expr\":\t{\n"
"\t\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\t\"Column\":\t{\n"
"\t\t\t\t\t\t\"Type\":\t9,\n"
"\t\t\t\t\t\t\"ColId\":\t1,\n"
"\t\t\t\t\t\t\"Bytes\":\t8\n"
"\t\t\t\t\t}\n"
"\t\t\t\t}\n"
"\t\t\t}, {\n"
"\t\t\t\t\"Base\":\t{\n"
"\t\t\t\t\t\"Schema\":\t{\n"
"\t\t\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\t\t\"ColId\":\t5001,\n"
"\t\t\t\t\t\t\"Bytes\":\t4\n"
"\t\t\t\t\t},\n"
"\t\t\t\t\t\"Columns\":\t[{\n"
"\t\t\t\t\t\t\t\"TableId\":\t1,\n"
"\t\t\t\t\t\t\t\"Flag\":\t0,\n"
"\t\t\t\t\t\t\t\"Info\":\t{\n"
"\t\t\t\t\t\t\t\t\"ColId\":\t2,\n"
"\t\t\t\t\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\t\t\t\t\"Bytes\":\t4\n"
"\t\t\t\t\t\t\t}\n"
"\t\t\t\t\t\t}],\n"
"\t\t\t\t\t\"InterBytes\":\t0\n"
"\t\t\t\t},\n"
"\t\t\t\t\"Expr\":\t{\n"
"\t\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\t\"Column\":\t{\n"
"\t\t\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\t\t\"ColId\":\t2,\n"
"\t\t\t\t\t\t\"Bytes\":\t4\n"
"\t\t\t\t\t}\n"
"\t\t\t\t}\n"
"\t\t\t}],\n"
"\t\t\"InputSchema\":\t[{\n"
"\t\t\t\t\"Type\":\t9,\n"
"\t\t\t\t\"ColId\":\t5000,\n"
"\t\t\t\t\"Bytes\":\t8\n"
"\t\t\t}, {\n"
"\t\t\t\t\"Type\":\t4,\n"
"\t\t\t\t\"ColId\":\t5001,\n"
"\t\t\t\t\"Bytes\":\t4\n"
"\t\t\t}],\n"
"\t\t\"TableScan\":\t{\n"
"\t\t\t\"TableId\":\t1,\n"
"\t\t\t\"TableType\":\t2,\n"
"\t\t\t\"Flag\":\t0,\n"
"\t\t\t\"Window\":\t{\n"
"\t\t\t\t\"StartKey\":\t-9.2233720368547758e+18,\n"
"\t\t\t\t\"EndKey\":\t9.2233720368547758e+18\n"
"\t\t\t}\n"
"\t\t}\n"
"\t},\n"
"\t\"DataSink\":\t{\n"
"\t\t\"Name\":\t\"Dispatch\",\n"
"\t\t\"Dispatch\":\t{\n"
"\t\t}\n"
"\t}\n"
"}";
SExecTaskInfo* pTaskInfo = nullptr;
int32_t code = qCreateExecTask((void*) 1, 2, NULL, (void**) &pTaskInfo);
}

3
source/libs/parser/src/parser.c
View File

@ -70,6 +70,9 @@ int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
int32_t code = qParserValidateSqlNode(&pCxt->ctx, &info, pQueryInfo, pCxt->pMsg, pCxt->msgLen);
if (code == TSDB_CODE_SUCCESS) {
*pQuery = (SQueryNode*)pQueryInfo;
} else {
terrno = code;
return code;
}
}

2
source/libs/planner/inc/plannerInt.h
View File

@ -62,7 +62,7 @@ typedef struct SQueryPlanNode {
SSchema *pSchema; // the schema of the input SSDatablock
int32_t numOfCols; // number of input columns
SArray *pExpr; // the query functions or sql aggregations
int32_t numOfExpr; // number of result columns, which is also the number of pExprs
int32_t numOfExpr; // number of result columns, which is also the number of pExprs
void *pExtInfo; // additional information
// children operator to generated result for current node to process
// in case of join, multiple prev nodes exist.

11
source/libs/planner/src/physicalPlan.c
View File

@ -88,16 +88,20 @@ static bool copySchema(SDataBlockSchema* dst, const SDataBlockSchema* src) {
}
static bool toDataBlockSchema(SQueryPlanNode* pPlanNode, SDataBlockSchema* dataBlockSchema) {
dataBlockSchema->numOfCols = pPlanNode->numOfCols;
dataBlockSchema->numOfCols = pPlanNode->numOfExpr;
dataBlockSchema->pSchema = malloc(sizeof(SSlotSchema) * pPlanNode->numOfCols);
if (NULL == dataBlockSchema->pSchema) {
return false;
}
memcpy(dataBlockSchema->pSchema, pPlanNode->pSchema, sizeof(SSlotSchema) * pPlanNode->numOfCols);
dataBlockSchema->resultRowSize = 0;
for (int32_t i = 0; i < dataBlockSchema->numOfCols; ++i) {
for (int32_t i = 0; i < pPlanNode->numOfExpr; ++i) {
SExprInfo* pExprInfo = taosArrayGetP(pPlanNode->pExpr, i);
memcpy(&dataBlockSchema->pSchema[i], &pExprInfo->base.resSchema, sizeof(SSlotSchema));
dataBlockSchema->resultRowSize += dataBlockSchema->pSchema[i].bytes;
}
return true;
}
@ -284,7 +288,6 @@ static SPhyNode* createSingleTableScanNode(SQueryPlanNode* pPlanNode, SQueryTabl
return createUserTableScanNode(pPlanNode, pTable, OP_TableScan);
}
static SPhyNode* createTableScanNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
SQueryTableInfo* pTable = (SQueryTableInfo*)pPlanNode->pExtInfo;

13
source/libs/planner/src/physicalPlanJson.c
View File

@ -154,9 +154,13 @@ static bool fromRawArrayWithAlloc(const cJSON* json, const char* name, FFromJson
return fromItem(jArray, func, *array, itemSize, *size);
}
static bool fromRawArray(const cJSON* json, const char* name, FFromJson func, void* array, int32_t itemSize, int32_t* size) {
static bool fromRawArray(const cJSON* json, const char* name, FFromJson func, void** array, int32_t itemSize, int32_t* size) {
const cJSON* jArray = getArray(json, name, size);
return fromItem(jArray, func, array, itemSize, *size);
if (*array == NULL) {
*array = calloc(*size, itemSize);
}
return fromItem(jArray, func, *array, itemSize, *size);
}
static char* getString(const cJSON* json, const char* name) {
@ -218,7 +222,8 @@ static bool dataBlockSchemaFromJson(const cJSON* json, void* obj) {
SDataBlockSchema* schema = (SDataBlockSchema*)obj;
schema->resultRowSize = getNumber(json, jkDataBlockSchemaResultRowSize);
schema->precision = getNumber(json, jkDataBlockSchemaPrecision);
return fromRawArray(json, jkDataBlockSchemaSlotSchema, schemaFromJson, schema->pSchema, sizeof(SSlotSchema), &schema->numOfCols);
return fromRawArray(json, jkDataBlockSchemaSlotSchema, schemaFromJson, (void**) &(schema->pSchema), sizeof(SSlotSchema), &schema->numOfCols);
}
static const char* jkColumnFilterInfoLowerRelOptr = "LowerRelOptr";
@ -920,6 +925,8 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
}
*str = cJSON_Print(json);
printf("%s\n", *str);
*len = strlen(*str) + 1;
return TSDB_CODE_SUCCESS;
}

36
source/libs/planner/src/planner.c
View File

@ -56,35 +56,43 @@ void qDestroyQueryDag(struct SQueryDag* pDag) {
tfree(pDag);
}
int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, uint64_t requestId) {
SQueryPlanNode* logicPlan;
int32_t code = createQueryPlan(pNode, &logicPlan);
int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pSchema, uint32_t* numOfResCols, uint64_t requestId) {
SQueryPlanNode* pLogicPlan;
int32_t code = createQueryPlan(pNode, &pLogicPlan);
if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(logicPlan);
destroyQueryPlan(pLogicPlan);
return code;
}
//
if (logicPlan->info.type != QNODE_MODIFY) {
// char* str = NULL;
// queryPlanToString(logicPlan, &str);
// printf("%s\n", str);
if (pLogicPlan->info.type != QNODE_MODIFY) {
char* str = NULL;
queryPlanToString(pLogicPlan, &str);
printf("%s\n", str);
}
code = optimizeQueryPlan(logicPlan);
int32_t numOfOutput = pLogicPlan->numOfExpr;
*pSchema = calloc(numOfOutput, sizeof(SSchema));
*numOfResCols = numOfOutput;
for(int32_t i = 0; i < numOfOutput; ++i) {
SExprInfo* pExprInfo = taosArrayGetP(pLogicPlan->pExpr, i);
memcpy(&(*pSchema)[i], pExprInfo->pExpr->pSchema, sizeof(SSchema));
}
code = optimizeQueryPlan(pLogicPlan);
if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(logicPlan);
destroyQueryPlan(pLogicPlan);
return code;
}
code = createDag(logicPlan, NULL, pDag, requestId);
code = createDag(pLogicPlan, NULL, pDag, requestId);
if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(logicPlan);
destroyQueryPlan(pLogicPlan);
qDestroyQueryDag(*pDag);
return code;
}
destroyQueryPlan(logicPlan);
destroyQueryPlan(pLogicPlan);
return TSDB_CODE_SUCCESS;
}

5
source/libs/planner/test/phyPlanTests.cpp
View File

@ -62,7 +62,10 @@ protected:
}
SQueryDag* dag = nullptr;
uint64_t requestId = 20;
code = qCreateQueryDag(query, &dag, requestId);
SSchema *schema = NULL;
uint32_t numOfOutput = 0;
code = qCreateQueryDag(query, &dag, &schema, &numOfOutput, requestId);
dag_.reset(dag);
return code;
}

26
source/libs/qworker/CMakeLists.txt
View File

@ -1,15 +1,31 @@
aux_source_directory(src QWORKER_SRC)
add_library(qworker ${QWORKER_SRC})
#add_library(qworker ${QWORKER_SRC})
#target_include_directories(
# qworker
# PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/qworker"
# PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
#)
#
#target_link_libraries(
# qworker
# PRIVATE os util transport planner qcom executor
#)
add_library(qworker STATIC ${QWORKER_SRC})
target_include_directories(
qworker
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/qworker"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
qworker
PRIVATE os util transport planner qcom
)
#set_target_properties(qworker PROPERTIES
# IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/libqworker.a"
# INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_SOURCE_DIR}/include/libs/qworker"
# )
target_link_libraries(qworker
PRIVATE os util transport planner qcom executor
)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)

13
source/libs/qworker/inc/qworkerInt.h
View File

@ -67,15 +67,16 @@ typedef struct SQWTaskStatus {
bool drop;
} SQWTaskStatus;
typedef struct SQWorkerResCache {
SRWLatch lock;
void *data;
} SQWorkerResCache;
typedef struct SQWorkerTaskHandleCache {
SRWLatch lock;
qTaskInfo_t taskHandle;
DataSinkHandle sinkHandle;
} SQWorkerTaskHandleCache;
typedef struct SQWSchStatus {
int32_t lastAccessTs; // timestamp in second
SRWLatch tasksLock;
SHashObj *tasksHash; // key:queryId+taskId, value: SQWorkerTaskStatus
SHashObj *tasksHash; // key:queryId+taskId, value: SQWTaskStatus
} SQWSchStatus;
// Qnode/Vnode level task management
@ -83,7 +84,7 @@ typedef struct SQWorkerMgmt {
SQWorkerCfg cfg;
SRWLatch schLock;
SRWLatch resLock;
SHashObj *schHash; //key: schedulerId, value: SQWorkerSchStatus
SHashObj *schHash; //key: schedulerId, value: SQWSchStatus
SHashObj *resHash; //key: queryId+taskId, value: SQWorkerResCache
} SQWorkerMgmt;

28
source/libs/qworker/src/qworker.c
View File

@ -1,9 +1,11 @@
#include "qworker.h"
#include "tname.h"
#include <common.h>
#include "executor.h"
#include "planner.h"
#include "query.h"
#include "qworkerInt.h"
#include "tmsg.h"
#include "tname.h"
int32_t qwValidateStatus(int8_t oriStatus, int8_t newStatus) {
int32_t code = 0;
@ -89,12 +91,13 @@ int32_t qwUpdateTaskInfo(SQWTaskStatus *task, int8_t type, void *data) {
return TSDB_CODE_SUCCESS;
}
int32_t qwAddTaskResCache(SQWorkerMgmt *mgmt, uint64_t qId, uint64_t tId, void *data) {
int32_t qwAddTaskAndSinkToCache(SQWorkerMgmt *mgmt, uint64_t qId, uint64_t tId, qTaskInfo_t taskHandle, DataSinkHandle sinkHandle) {
char id[sizeof(qId) + sizeof(tId)] = {0};
QW_SET_QTID(id, qId, tId);
SQWorkerResCache resCache = {0};
resCache.data = data;
resCache.taskHandle = taskHandle;
resCache.sinkHandle = sinkHandle;
QW_LOCK(QW_WRITE, &mgmt->resLock);
if (0 != taosHashPut(mgmt->resHash, id, sizeof(id), &resCache, sizeof(SQWorkerResCache))) {
@ -1011,7 +1014,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
bool queryDone = false;
bool queryRsped = false;
bool needStop = false;
SSubplan *plan = NULL;
struct SSubplan *plan = NULL;
QW_ERR_JRET(qwCheckTaskCancelDrop(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId, &needStop));
if (needStop) {
@ -1025,10 +1028,8 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
QW_ERR_JRET(code);
}
//TODO call executer to init subquery
code = 0; // return error directly
//TODO call executer to init subquery
qTaskInfo_t pTaskInfo = NULL;
code = qCreateExecTask(node, 0, (struct SSubplan *)plan, &pTaskInfo);
if (code) {
QW_ERR_JRET(code);
} else {
@ -1039,22 +1040,19 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
queryRsped = true;
//TODO call executer to execute subquery
code = 0;
void *data = NULL;
SSDataBlock* pRes = NULL;
code = qExecTask(pTaskInfo, &pRes);
queryDone = false;
//TODO call executer to execute subquery
if (code) {
QW_ERR_JRET(code);
} else {
QW_ERR_JRET(qwAddTaskResCache(qWorkerMgmt, msg->queryId, msg->taskId, data));
QW_ERR_JRET(qwAddTaskAndSinkToCache(qWorkerMgmt, msg->queryId, msg->taskId, pTaskInfo, NULL));
QW_ERR_JRET(qwUpdateTaskStatus(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId, JOB_TASK_STATUS_PARTIAL_SUCCEED));
}
_return:
if (queryRsped) {
code = qwCheckAndSendReadyRsp(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId, pMsg, code);
} else {

16
source/libs/scheduler/src/scheduler.c
View File

@ -269,7 +269,6 @@ int32_t schRecordTaskExecNode(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *ad
int32_t schValidateAndBuildJob(SQueryDag *pDag, SSchJob *pJob) {
int32_t code = 0;
pJob->queryId = pDag->queryId;
if (pDag->numOfSubplans <= 0) {
@ -367,7 +366,6 @@ int32_t schValidateAndBuildJob(SQueryDag *pDag, SSchJob *pJob) {
SCH_ERR_JRET(schBuildTaskRalation(pJob, planToTask));
_return:
if (planToTask) {
taosHashCleanup(planToTask);
}
@ -762,6 +760,8 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
break;
}
break;
}
case TDMT_VND_QUERY_RSP: {
SQueryTableRsp *rsp = (SQueryTableRsp *)msg;
@ -773,6 +773,8 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
break;
}
break;
}
case TDMT_VND_RES_READY_RSP: {
SResReadyRsp *rsp = (SResReadyRsp *)msg;
@ -784,6 +786,8 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
break;
}
break;
}
case TDMT_VND_FETCH_RSP: {
SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
@ -1325,7 +1329,7 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void
}
int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob) {
if (NULL == transport || /*NULL == nodeList || */NULL == pDag || NULL == pDag->pSubplans || NULL == pJob) {
if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob) {
SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
@ -1485,8 +1489,4 @@ void scheduleFreeJob(void *job) {
void schedulerDestroy(void) {
if (schMgmt.jobs) {
taosHashCleanup(schMgmt.jobs); //TODO
schMgmt.jobs = NULL;
}
}
taosHashCleanup(sch