Skip to content

T
TDengine

buyaopa / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 3
Star 0
Fork 0
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 ecae31bb 编写于 作者: S slguan
浏览文件
操作

Merge branch '2.0' into refact/slguan 

# Conflicts:
#	src/vnode/detail/src/vnodeUtil.c
上级 86e4ebcd 00092835
展开全部 隐藏空白更改
内联 并排
Showing with 5478 addition and 1558 deletion
src/client/inc/tscUtil.h
浏览文件 @ ecae31bb
......@@ -32,7 +32,7 @@ extern "C" {
#define UTIL_TABLE_IS_SUPERTABLE(metaInfo) \
(((metaInfo)->pTableMeta != NULL) && ((metaInfo)->pTableMeta->tableType == TSDB_SUPER_TABLE))
#define UTIL_TABLE_IS_NOMRAL_TABLE(metaInfo) (!(UTIL_TABLE_IS_SUPERTABLE(metaInfo)))
#define UTIL_TABLE_CREATE_FROM_STABLE(metaInfo) \
#define UTIL_TABLE_IS_CHILD_TABLE(metaInfo) \
(((metaInfo)->pTableMeta != NULL) && ((metaInfo)->pTableMeta->tableType == TSDB_CHILD_TABLE))
#define TSDB_COL_IS_TAG(f) (((f)&TSDB_COL_TAG) != 0)
......@@ -106,7 +106,7 @@ bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo);
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscQueryOnMetric(SSqlCmd* pCmd);
bool tscQueryMetricTags(SQueryInfo* pQueryInfo);
bool tscQueryTags(SQueryInfo* pQueryInfo);
bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd);
void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex, int16_t functionId, SColumnIndex* pIndex,
......@@ -176,8 +176,8 @@ void tscIncStreamExecutionCount(void* pStream);
bool tscValidateColumnId(STableMetaInfo* pTableMetaInfo, int32_t colId);
// get starter position of metric query condition (query on tags) in SSqlCmd.payload
SCond* tsGetMetricQueryCondPos(STagCond* pCond, uint64_t tableIndex);
void tsSetMetricQueryCond(STagCond* pTagCond, uint64_t uid, const char* str);
SCond* tsGetSTableQueryCondPos(STagCond* pCond, uint64_t tableIndex);
void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, const char* str);
void tscTagCondCopy(STagCond* dest, const STagCond* src);
void tscTagCondRelease(STagCond* pCond);
......@@ -207,7 +207,7 @@ void tscFreeSubqueryInfo(SSqlCmd* pCmd);
void tscClearSubqueryInfo(SSqlCmd* pCmd);
void tscGetMetricMetaCacheKey(SQueryInfo* pQueryInfo, char* keyStr, uint64_t uid);
int tscGetMetricMeta(SSqlObj* pSql, int32_t clauseIndex);
int tscGetSTableVgroupInfo(SSqlObj* pSql, int32_t clauseIndex);
int tscGetTableMeta(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo);
int tscGetMeterMetaEx(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, bool createIfNotExists);
......
src/client/inc/tsclient.h
浏览文件 @ ecae31bb
......@@ -70,8 +70,9 @@ typedef struct STableMeta {
typedef struct STableMetaInfo {
STableMeta * pTableMeta; // table meta, cached in client side and acquried by name
SSuperTableMeta *pMetricMeta; // metricmeta
// SSuperTableMeta *pMetricMeta; // metricmeta
SArray* vgroupIdList;
/*
* 1. keep the vnode index during the multi-vnode super table projection query
* 2. keep the vnode index for multi-vnode insertion
......
src/client/src/tscAsync.c
浏览文件 @ ecae31bb
......@@ -406,11 +406,6 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
if (pSql->fp == NULL) {
tscError("%p callBack is NULL!!!", pSql);
return;
}
if (pSql->fp == (void *)1) {
pSql->fp = NULL;
......@@ -464,7 +459,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
code = tscGetMetricMeta(pSql, 0);
code = tscGetSTableVgroupInfo(pSql, 0);
pRes->code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
......@@ -494,7 +489,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
if (code == TSDB_CODE_SUCCESS && UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
code = tscGetMetricMeta(pSql, pCmd->clauseIndex);
code = tscGetSTableVgroupInfo(pSql, pCmd->clauseIndex);
pRes->code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
......@@ -519,7 +514,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
tscIncStreamExecutionCount(pSql->pStream);
} else {
tscTrace("%p get tableMeta/metricMeta successfully", pSql);
tscTrace("%p get tableMeta successfully", pSql);
}
tscDoQuery(pSql);
......
src/client/src/tscLocal.c
浏览文件 @ ecae31bb
......@@ -286,6 +286,7 @@ static int32_t tscProcessDescribeTable(SSqlObj *pSql) {
// todo add order support
static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
#if 0
// the result structure has been completed in sql parse, so we
// only need to reorganize the results in the column format
SSqlCmd * pCmd = &pSql->cmd;
......@@ -337,6 +338,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
}
}
#endif
return 0;
}
......@@ -345,7 +347,7 @@ static int tscBuildMetricTagSqlFunctionResult(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
#if 0
SSuperTableMeta *pMetricMeta = tscGetMetaInfo(pQueryInfo, 0)->pMetricMeta;
int32_t totalNumOfResults = 1; // count function only produce one result
int32_t rowLen = tscGetResRowLength(pQueryInfo);
......@@ -369,7 +371,8 @@ static int tscBuildMetricTagSqlFunctionResult(SSqlObj *pSql) {
}
rowIdx++;
}
#endif
return 0;
}
......
src/client/src/tscParseInsert.c
浏览文件 @ ecae31bb
......@@ -1238,8 +1238,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
goto _clean;
}
if (pCmd->pDataBlocks->nSize > 0) {
// merge according to vgId
if (pCmd->pDataBlocks->nSize > 0) { // merge according to vgId
if ((code = tscMergeTableDataBlocks(pSql, pCmd->pDataBlocks)) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
......@@ -1294,12 +1293,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion) {
int32_t ret = TSDB_CODE_SUCCESS;
if (NULL == pSql->asyncTblPos) {
tscCleanSqlCmd(&pSql->cmd);
} else {
tscTrace("continue parse sql: %s", pSql->asyncTblPos);
}
tscTrace("continue parse sql: %s", pSql->asyncTblPos);
if (tscIsInsertOrImportData(pSql->sqlstr)) {
/*
......
src/client/src/tscSQLParser.c
浏览文件 @ ecae31bb
......@@ -209,9 +209,14 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
}
int32_t code = tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex, &pQueryInfo);
assert(pQueryInfo->numOfTables == 0);
// assert(pQueryInfo->numOfTables == 0);
STableMetaInfo* pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
STableMetaInfo* pTableMetaInfo = NULL;
if (pQueryInfo->numOfTables == 0) {
pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
} else {
pTableMetaInfo = &pQueryInfo->pTableMetaInfo[0];
}
pCmd->command = pInfo->type;
......@@ -639,7 +644,7 @@ int32_t parseIntervalClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
* check invalid SQL:
* select tbname, tags_fields from super_table_name interval(1s)
*/
if (tscQueryMetricTags(pQueryInfo) && pQueryInfo->intervalTime > 0) {
if (tscQueryTags(pQueryInfo) && pQueryInfo->intervalTime > 0) {
return invalidSqlErrMsg(pQueryInfo->msg, msg1);
}
......@@ -746,7 +751,7 @@ int32_t setMeterID(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SSqlO
tscClearMeterMetaInfo(pTableMetaInfo, false);
}
} else {
assert(pTableMetaInfo->pTableMeta == NULL && pTableMetaInfo->pMetricMeta == NULL);
assert(pTableMetaInfo->pTableMeta == NULL);
}
tfree(oldName);
......@@ -1252,7 +1257,7 @@ int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, tSQLExprList* pSel
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
int32_t numOfCols = tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
if (tscQueryMetricTags(pQueryInfo)) { // local handle the metric tag query
if (tscQueryTags(pQueryInfo)) { // local handle the metric tag query
pCmd->count = numOfCols; // the number of meter schema, tricky.
pQueryInfo->command = TSDB_SQL_RETRIEVE_TAGS;
}
......@@ -1293,7 +1298,7 @@ SSqlExpr* doAddProjectCol(SQueryInfo* pQueryInfo, int32_t outputIndex, int32_t c
int16_t functionId = (int16_t)((colIdx >= numOfCols) ? TSDB_FUNC_TAGPRJ : TSDB_FUNC_PRJ);
if (functionId == TSDB_FUNC_TAGPRJ) {
addRequiredTagColumn(pQueryInfo, colIdx - numOfCols, tableIndex);
// addRequiredTagColumn(pQueryInfo, colIdx - numOfCols, tableIndex);
pQueryInfo->type = TSDB_QUERY_TYPE_STABLE_QUERY;
} else {
pQueryInfo->type = TSDB_QUERY_TYPE_PROJECTION_QUERY;
......@@ -1396,8 +1401,7 @@ static int32_t doAddProjectionExprAndResultFields(SQueryInfo* pQueryInfo, SColum
SColumnList ids = {0};
ids.ids[0] = *pIndex;
// tag columns do not add to source list
ids.num = (j >= tscGetNumOfColumns(pTableMeta)) ? 0 : 1;
ids.num = 1;
insertResultField(pQueryInfo, startPos + j, &ids, pSchema[j].bytes, pSchema[j].type, pSchema[j].name, pExpr);
}
......@@ -2785,7 +2789,7 @@ static int32_t optrToString(tSQLExpr* pExpr, char** exprString) {
return TSDB_CODE_SUCCESS;
}
static int32_t tablenameListToString(tSQLExpr* pExpr, /*char* str*/ SStringBuilder* sb) {
static int32_t tablenameListToString(tSQLExpr* pExpr, SStringBuilder* sb) {
tSQLExprList* pList = pExpr->pParam;
if (pList->nExpr <= 0) {
return TSDB_CODE_INVALID_SQL;
......@@ -2811,7 +2815,7 @@ static int32_t tablenameListToString(tSQLExpr* pExpr, /*char* str*/ SStringBuild
return TSDB_CODE_SUCCESS;
}
static int32_t tablenameCondToString(tSQLExpr* pExpr, /*char* str*/ SStringBuilder* sb) {
static int32_t tablenameCondToString(tSQLExpr* pExpr, SStringBuilder* sb) {
taosStringBuilderAppendStringLen(sb, QUERY_COND_REL_PREFIX_LIKE, QUERY_COND_REL_PREFIX_LIKE_LEN);
taosStringBuilderAppendString(sb, pExpr->val.pz);
......@@ -3507,7 +3511,7 @@ int tableNameCompar(const void* lhs, const void* rhs) {
return ret > 0 ? 1 : -1;
}
static int32_t setTableCondForMetricQuery(SQueryInfo* pQueryInfo, const char* account, tSQLExpr* pExpr,
static int32_t setTableCondForSTableQuery(SQueryInfo* pQueryInfo, const char* account, tSQLExpr* pExpr,
int16_t tableCondIndex, SStringBuilder* sb) {
const char* msg = "table name too long";
......@@ -3736,7 +3740,7 @@ static int32_t getTagQueryCondExpr(SQueryInfo* pQueryInfo, SCondExpr* pCondExpr,
return ret;
}
tsSetMetricQueryCond(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid, c);
tsSetSTableQueryCond(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid, c);
doCompactQueryExpr(pExpr);
tSQLExprDestroy(p1);
......@@ -3752,8 +3756,8 @@ int32_t parseWhereClause(SQueryInfo* pQueryInfo, tSQLExpr** pExpr, SSqlObj* pSql
return TSDB_CODE_SUCCESS;
}
const char* msg = "invalid filter expression";
const char* msg1 = "invalid expression";
const char* msg2 = "invalid filter expression";
int32_t ret = TSDB_CODE_SUCCESS;
......@@ -3811,11 +3815,11 @@ int32_t parseWhereClause(SQueryInfo* pQueryInfo, tSQLExpr** pExpr, SSqlObj* pSql
// 7. query condition for table name
pQueryInfo->tagCond.relType = (condExpr.relType == TK_AND) ? TSDB_RELATION_AND : TSDB_RELATION_OR;
ret = setTableCondForMetricQuery(pQueryInfo, getAccountId(pSql), condExpr.pTableCond, condExpr.tableCondIndex, &sb);
ret = setTableCondForSTableQuery(pQueryInfo, getAccountId(pSql), condExpr.pTableCond, condExpr.tableCondIndex, &sb);
taosStringBuilderDestroy(&sb);
if (!validateFilterExpr(pQueryInfo)) {
return invalidSqlErrMsg(pQueryInfo->msg, msg);
return invalidSqlErrMsg(pQueryInfo->msg, msg2);
}
doAddJoinTagsColumnsIntoTagList(pQueryInfo, &condExpr);
......@@ -4240,7 +4244,7 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
} else if ((pAlterSQL->type == TSDB_ALTER_TABLE_UPDATE_TAG_VAL) && (UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo))) {
return invalidSqlErrMsg(pQueryInfo->msg, msg4);
} else if ((pAlterSQL->type == TSDB_ALTER_TABLE_ADD_COLUMN || pAlterSQL->type == TSDB_ALTER_TABLE_DROP_COLUMN) &&
UTIL_TABLE_CREATE_FROM_STABLE(pTableMetaInfo)) {
UTIL_TABLE_IS_CHILD_TABLE(pTableMetaInfo)) {
return invalidSqlErrMsg(pQueryInfo->msg, msg6);
}
......@@ -4666,14 +4670,13 @@ int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t clauseIndex, SQuerySQL*
* And then launching multiple async-queries against all qualified virtual nodes, during the first-stage
* query operation.
*/
int32_t code = tscGetMetricMeta(pSql, clauseIndex);
int32_t code = tscGetSTableVgroupInfo(pSql, clauseIndex);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
// No tables included. No results generated. Query results are empty.
SSuperTableMeta* pMetricMeta = pTableMetaInfo->pMetricMeta;
if (pTableMetaInfo->pTableMeta == NULL || pMetricMeta == NULL || pMetricMeta->numOfTables == 0) {
if (pTableMetaInfo->pTableMeta == NULL) {
tscTrace("%p no table in metricmeta, no output result", pSql);
pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
}
......@@ -5441,8 +5444,9 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
// two table: the first one is for current table, and the secondary is for the super table.
tscAddEmptyMetaInfo(pQueryInfo);
assert(pQueryInfo->numOfTables == 2);
if (pQueryInfo->numOfTables < 2) {
tscAddEmptyMetaInfo(pQueryInfo);
}
const int32_t TABLE_INDEX = 0;
const int32_t STABLE_INDEX = 1;
......@@ -5687,6 +5691,13 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
}
assert(pQueryInfo->numOfTables == pQuerySql->from->nExpr);
if (UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
int32_t code = tscGetSTableVgroupInfo(pSql, index);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
// parse the group by clause in the first place
if (parseGroupbyClause(pQueryInfo, pQuerySql->pGroupby, pCmd) != TSDB_CODE_SUCCESS) {
......
src/client/src/tscSchemaUtil.c
浏览文件 @ ecae31bb
......@@ -32,7 +32,7 @@ int32_t tscGetNumOfTags(const STableMeta* pTableMeta) {
}
if (pTableMeta->tableType == TSDB_SUPER_TABLE || pTableMeta->tableType == TSDB_CHILD_TABLE) {
assert(tinfo.numOfTags > 0);
assert(tinfo.numOfTags >= 0);
return tinfo.numOfTags;
}
......@@ -51,12 +51,12 @@ int32_t tscGetNumOfColumns(const STableMeta* pTableMeta) {
SSchema *tscGetTableSchema(const STableMeta *pTableMeta) {
assert(pTableMeta != NULL);
if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
STableMeta* pSTableMeta = pTableMeta->pSTable;
assert (pSTableMeta != NULL);
return pSTableMeta->schema;
}
// if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
// STableMeta* pSTableMeta = pTableMeta->pSTable;
// assert (pSTableMeta != NULL);
//
// return pSTableMeta->schema;
// }
return (SSchema*) pTableMeta->schema;
}
......@@ -72,12 +72,14 @@ SSchema* tscGetTableTagSchema(const STableMeta* pTableMeta) {
STableComInfo tscGetTableInfo(const STableMeta* pTableMeta) {
assert(pTableMeta != NULL);
#if 0
if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
assert (pTableMeta->pSTable != NULL);
return pTableMeta->pSTable->tableInfo;
}
#endif
return pTableMeta->tableInfo;
}
......@@ -130,12 +132,13 @@ SSchema* tscGetTableColumnSchema(const STableMeta* pTableMeta, int32_t startCol)
assert(pTableMeta != NULL);
SSchema* pSchema = pTableMeta->schema;
#if 0
if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
assert (pTableMeta->pSTable != NULL);
pSchema = pTableMeta->pSTable->schema;
}
#endif
return &pSchema[startCol];
}
......
src/client/src/tscSecondaryMerge.c
浏览文件 @ ecae31bb
......@@ -605,7 +605,7 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
(*pMemBuffer) = (tExtMemBuffer **)malloc(POINTER_BYTES * pTableMetaInfo->pMetricMeta->numOfVnodes);
(*pMemBuffer) = (tExtMemBuffer **)malloc(POINTER_BYTES * 1);
if (*pMemBuffer == NULL) {
tscError("%p failed to allocate memory", pSql);
pRes->code = TSDB_CODE_CLI_OUT_OF_MEMORY;
......@@ -636,7 +636,8 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
pModel = createColumnModel(pSchema, pQueryInfo->exprsInfo.numOfExprs, capacity);
for (int32_t i = 0; i < pTableMetaInfo->pMetricMeta->numOfVnodes; ++i) {
size_t numOfSubs = taosArrayGetSize(pTableMetaInfo->vgroupIdList);
for (int32_t i = 0; i < numOfSubs; ++i) {
(*pMemBuffer)[i] = createExtMemBuffer(nBufferSizes, rlen, pModel);
(*pMemBuffer)[i]->flushModel = MULTIPLE_APPEND_MODEL;
}
......
src/client/src/tscServer.c
浏览文件 @ ecae31bb
此差异已折叠。
src/client/src/tscSql.c
浏览文件 @ ecae31bb
......@@ -546,11 +546,11 @@ static bool tscHashRemainDataInSubqueryResultSet(SSqlObj *pSql) {
* if the global limitation is not reached, and current result has not exhausted, or next more vnodes are
* available, goes on
*/
if (pMetaInfo->vnodeIndex < pMetaInfo->pMetricMeta->numOfVnodes && pRes1->row < pRes1->numOfRows &&
(!tscHasReachLimitation(pQueryInfo1, pRes1))) {
allSubqueryExhausted = false;
break;
}
// if (pMetaInfo->vnodeIndex < pMetaInfo->pMetricMeta->numOfVnodes && pRes1->row < pRes1->numOfRows &&
// (!tscHasReachLimitation(pQueryInfo1, pRes1))) {
// allSubqueryExhausted = false;
// break;
// }
}
hasData = !allSubqueryExhausted;
......@@ -651,8 +651,7 @@ static void **tscBuildResFromSubqueries(SSqlObj *pSql) {
static void asyncFetchCallback(void *param, TAOS_RES *tres, int numOfRows) {
SSqlObj* pSql = (SSqlObj*) tres;
if (numOfRows < 0) {
// set the error code
if (numOfRows < 0) { // set the error code
pSql->res.code = -numOfRows;
}
......
src/client/src/tscStream.c
浏览文件 @ ecae31bb
......@@ -80,7 +80,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
if (code == 0 && UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
code = tscGetMetricMeta(pSql, 0);
code = tscGetSTableVgroupInfo(pSql, 0);
pSql->res.code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
......
src/client/src/tscSub.c
浏览文件 @ ecae31bb
......@@ -178,11 +178,11 @@ int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, 0, 0);
int numOfTables = 0;
if (!UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
SSuperTableMeta* pMetricMeta = pTableMetaInfo->pMetricMeta;
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
numOfTables += pVnodeSidList->numOfSids;
}
// SSuperTableMeta* pMetricMeta = pTableMetaInfo->pMetricMeta;
// for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
// SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
// numOfTables += pVnodeSidList->numOfSids;
// }
}
SSubscriptionProgress* progress = (SSubscriptionProgress*)calloc(numOfTables, sizeof(SSubscriptionProgress));
......@@ -197,17 +197,17 @@ int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
progress[0].uid = uid;
progress[0].key = tscGetSubscriptionProgress(pSub, uid);
} else {
SSuperTableMeta* pMetricMeta = pTableMetaInfo->pMetricMeta;
numOfTables = 0;
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) {
STableIdInfo *pTableMetaInfo = tscGetMeterSidInfo(pVnodeSidList, j);
int64_t uid = pTableMetaInfo->uid;
progress[numOfTables].uid = uid;
progress[numOfTables++].key = tscGetSubscriptionProgress(pSub, uid);
}
}
// SSuperTableMeta* pMetricMeta = pTableMetaInfo->pMetricMeta;
// numOfTables = 0;
// for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
// SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
// for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) {
// STableIdInfo *pTableMetaInfo = tscGetMeterSidInfo(pVnodeSidList, j);
// int64_t uid = pTableMetaInfo->uid;
// progress[numOfTables].uid = uid;
// progress[numOfTables++].key = tscGetSubscriptionProgress(pSub, uid);
// }
// }
qsort(progress, numOfTables, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress);
}
......
src/client/src/tscSubquery.c
浏览文件 @ ecae31bb
......@@ -476,7 +476,8 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
assert(pQueryInfo->numOfTables == 1);
// for projection query, need to try next vnode
int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
int32_t totalVnode = 0;
if ((++pTableMetaInfo->vnodeIndex) < totalVnode) {
tscTrace("%p current vnode:%d exhausted, try next:%d. total vnode:%d. current numOfRes:%d", pSql,
pTableMetaInfo->vnodeIndex - 1, pTableMetaInfo->vnodeIndex, totalVnode, pRes->numOfTotal);
......@@ -541,16 +542,16 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
assert(pQueryInfo->numOfTables == 1);
// for projection query, need to try next vnode if current vnode is exhausted
if ((++pTableMetaInfo->vnodeIndex) < pTableMetaInfo->pMetricMeta->numOfVnodes) {
pSupporter->pState->numOfCompleted = 0;
pSupporter->pState->numOfTotal = 1;
pSql->cmd.command = TSDB_SQL_SELECT;
pSql->fp = tscJoinQueryCallback;
tscProcessSql(pSql);
return;
}
// if ((++pTableMetaInfo->vnodeIndex) < pTableMetaInfo->pMetricMeta->numOfVnodes) {
// pSupporter->pState->numOfCompleted = 0;
// pSupporter->pState->numOfTotal = 1;
//
// pSql->cmd.command = TSDB_SQL_SELECT;
// pSql->fp = tscJoinQueryCallback;
// tscProcessSql(pSql);
//
// return;
// }
}
int32_t numOfTotal = pSupporter->pState->numOfTotal;
......@@ -608,10 +609,10 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) {
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
if (pRes->row >= pRes->numOfRows && pTableMetaInfo->vnodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes &&
(!tscHasReachLimitation(pQueryInfo, pRes))) {
numOfFetch++;
}
// if (pRes->row >= pRes->numOfRows && pTableMetaInfo->vnodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes &&
// (!tscHasReachLimitation(pQueryInfo, pRes))) {
// numOfFetch++;
// }
} else {
if (pRes->row >= pRes->numOfRows && (!tscHasReachLimitation(pQueryInfo, pRes))) {
numOfFetch++;
......@@ -788,7 +789,7 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
* data instead of returning to its invoker
*/
if (pTableMetaInfo->vnodeIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
assert(pTableMetaInfo->vnodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes);
// assert(pTableMetaInfo->vnodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes);
pSupporter->pState->numOfCompleted = 0; // reset the record value
pSql->fp = joinRetrieveCallback; // continue retrieve data
......@@ -1005,11 +1006,12 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
pRes->qhandle = 1; // hack the qhandle check
const uint32_t nBufferSize = (1 << 16); // 64KB
const uint32_t nBufferSize = (1u << 16); // 64KB
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
int32_t numOfSubQueries = pTableMetaInfo->pMetricMeta->numOfVnodes;
int32_t numOfSubQueries = taosArrayGetSize(pTableMetaInfo->vgroupIdList);
assert(numOfSubQueries > 0);
int32_t ret = tscLocalReducerEnvCreate(pSql, &pMemoryBuf, &pDesc, &pModel, nBufferSize);
......@@ -1115,7 +1117,7 @@ static void tscFreeSubSqlObj(SRetrieveSupport *trsupport, SSqlObj *pSql) {
tfree(trsupport);
}
static void tscRetrieveFromVnodeCallBack(void *param, TAOS_RES *tres, int numOfRows);
static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfRows);
static void tscAbortFurtherRetryRetrieval(SRetrieveSupport *trsupport, TAOS_RES *tres, int32_t errCode) {
// set no disk space error info
......@@ -1137,7 +1139,7 @@ static void tscAbortFurtherRetryRetrieval(SRetrieveSupport *trsupport, TAOS_RES
pthread_mutex_unlock(&trsupport->queryMutex);
tscRetrieveFromVnodeCallBack(trsupport, tres, trsupport->pState->code);
tscRetrieveFromDnodeCallBack(trsupport, tres, trsupport->pState->code);
}
static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numOfRows) {
......@@ -1232,7 +1234,86 @@ static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSq
}
}
static void tscRetrieveFromVnodeCallBack(void *param, TAOS_RES *tres, int numOfRows) {
static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* pSql) {
int32_t idx = trsupport->subqueryIndex;
SSqlObj * pPObj = trsupport->pParentSqlObj;
tOrderDescriptor *pDesc = trsupport->pOrderDescriptor;
SSubqueryState* pState = trsupport->pState;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
// data in from current vnode is stored in cache and disk
uint32_t numOfRowsFromSubquery = trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->numOfElems;
// tscTrace("%p sub:%p all data retrieved from ip:%u,vid:%d, numOfRows:%d, orderOfSub:%d", pPObj, pSql, pSvd->ip,
// pSvd->vnode, numOfRowsFromSubquery, idx);
tColModelCompact(pDesc->pColumnModel, trsupport->localBuffer, pDesc->pColumnModel->capacity);
#ifdef _DEBUG_VIEW
printf("%" PRIu64 " rows data flushed to disk:\n", trsupport->localBuffer->numOfElems);
SSrcColumnInfo colInfo[256] = {0};
tscGetSrcColumnInfo(colInfo, pQueryInfo);
tColModelDisplayEx(pDesc->pColumnModel, trsupport->localBuffer->data, trsupport->localBuffer->numOfElems,
trsupport->localBuffer->numOfElems, colInfo);
#endif
if (tsTotalTmpDirGB != 0 && tsAvailTmpDirGB < tsMinimalTmpDirGB) {
tscError("%p sub:%p client disk space remain %.3f GB, need at least %.3f GB, stop query", pPObj, pSql,
tsAvailTmpDirGB, tsMinimalTmpDirGB);
tscAbortFurtherRetryRetrieval(trsupport, pSql, TSDB_CODE_CLI_NO_DISKSPACE);
return;
}
// each result for a vnode is ordered as an independant list,
// then used as an input of loser tree for disk-based merge routine
int32_t ret = tscFlushTmpBuffer(trsupport->pExtMemBuffer[idx], pDesc, trsupport->localBuffer,
pQueryInfo->groupbyExpr.orderType);
if (ret != 0) { // set no disk space error info, and abort retry
return tscAbortFurtherRetryRetrieval(trsupport, pSql, TSDB_CODE_CLI_NO_DISKSPACE);
}
// keep this value local variable, since the pState variable may be released by other threads, if atomic_add opertion
// increases the finished value up to pState->numOfTotal value, which means all subqueries are completed.
// In this case, the comparsion between finished value and released pState->numOfTotal is not safe.
int32_t numOfTotal = pState->numOfTotal;
int32_t finished = atomic_add_fetch_32(&pState->numOfCompleted, 1);
if (finished < numOfTotal) {
tscTrace("%p sub:%p orderOfSub:%d freed, finished subqueries:%d", pPObj, pSql, trsupport->subqueryIndex, finished);
return tscFreeSubSqlObj(trsupport, pSql);
}
// all sub-queries are returned, start to local merge process
pDesc->pColumnModel->capacity = trsupport->pExtMemBuffer[idx]->numOfElemsPerPage;
tscTrace("%p retrieve from %d vnodes completed.final NumOfRows:%d,start to build loser tree", pPObj,
pState->numOfTotal, pState->numOfRetrievedRows);
SQueryInfo *pPQueryInfo = tscGetQueryInfoDetail(&pPObj->cmd, 0);
tscClearInterpInfo(pPQueryInfo);
tscCreateLocalReducer(trsupport->pExtMemBuffer, pState->numOfTotal, pDesc, trsupport->pFinalColModel,
&pPObj->cmd, &pPObj->res);
tscTrace("%p build loser tree completed", pPObj);
pPObj->res.precision = pSql->res.precision;
pPObj->res.numOfRows = 0;
pPObj->res.row = 0;
// only free once
tfree(trsupport->pState);
tscFreeSubSqlObj(trsupport, pSql);
// set the command flag must be after the semaphore been correctly set.
pPObj->cmd.command = TSDB_SQL_RETRIEVE_METRIC;
if (pPObj->res.code == TSDB_CODE_SUCCESS) {
(*pPObj->fp)(pPObj->param, pPObj, 0);
} else {
tscQueueAsyncRes(pPObj);
}
}
static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfRows) {
SRetrieveSupport *trsupport = (SRetrieveSupport *)param;
int32_t idx = trsupport->subqueryIndex;
SSqlObj * pPObj = trsupport->pParentSqlObj;
......@@ -1260,15 +1341,16 @@ static void tscRetrieveFromVnodeCallBack(void *param, TAOS_RES *tres, int numOfR
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
SVnodeSidList *vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
SVnodeDesc * pSvd = &vnodeInfo->vpeerDesc[vnodeInfo->index];
// SVnodeSidList *vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
// SVnodeSidList *vnodeInfo = 0;
// SVnodeDesc * pSvd = &vnodeInfo->vpeerDesc[vnodeInfo->index];
if (numOfRows > 0) {
assert(pRes->numOfRows == numOfRows);
int64_t num = atomic_add_fetch_64(&pState->numOfRetrievedRows, numOfRows);
tscTrace("%p sub:%p retrieve numOfRows:%d totalNumOfRows:%d from ip:%u,vid:%d,orderOfSub:%d", pPObj, pSql,
pRes->numOfRows, pState->numOfRetrievedRows, pSvd->ip, pSvd->vnode, idx);
// tscTrace("%p sub:%p retrieve numOfRows:%d totalNumOfRows:%d from ip:%u,vid:%d,orderOfSub:%d", pPObj, pSql,
// pRes->numOfRows, pState->numOfRetrievedRows, pSvd->ip, pSvd->vnode, idx);
if (num > tsMaxNumOfOrderedResults && tscIsProjectionQueryOnSTable(pQueryInfo, 0)) {
tscError("%p sub:%p num of OrderedRes is too many, max allowed:%" PRId64 " , current:%" PRId64,
......@@ -1294,85 +1376,16 @@ static void tscRetrieveFromVnodeCallBack(void *param, TAOS_RES *tres, int numOfR
int32_t ret = saveToBuffer(trsupport->pExtMemBuffer[idx], pDesc, trsupport->localBuffer, pRes->data,
pRes->numOfRows, pQueryInfo->groupbyExpr.orderType);
if (ret < 0) {
// set no disk space error info, and abort retry
if (ret < 0) { // set no disk space error info, and abort retry
tscAbortFurtherRetryRetrieval(trsupport, tres, TSDB_CODE_CLI_NO_DISKSPACE);
} else {
} else if (pRes->completed) {
tscAllDataRetrievedFromDnode(trsupport, pSql);
} else { // continue fetch data from dnode
pthread_mutex_unlock(&trsupport->queryMutex);
taos_fetch_rows_a(tres, tscRetrieveFromVnodeCallBack, param);
}
} else { // all data has been retrieved to client
/* data in from current vnode is stored in cache and disk */
uint32_t numOfRowsFromVnode = trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->numOfElems;
tscTrace("%p sub:%p all data retrieved from ip:%u,vid:%d, numOfRows:%d, orderOfSub:%d", pPObj, pSql, pSvd->ip,
pSvd->vnode, numOfRowsFromVnode, idx);
tColModelCompact(pDesc->pColumnModel, trsupport->localBuffer, pDesc->pColumnModel->capacity);
#ifdef _DEBUG_VIEW
printf("%" PRIu64 " rows data flushed to disk:\n", trsupport->localBuffer->numOfElems);
SSrcColumnInfo colInfo[256] = {0};
tscGetSrcColumnInfo(colInfo, pQueryInfo);
tColModelDisplayEx(pDesc->pColumnModel, trsupport->localBuffer->data, trsupport->localBuffer->numOfElems,
trsupport->localBuffer->numOfElems, colInfo);
#endif
if (tsTotalTmpDirGB != 0 && tsAvailTmpDirGB < tsMinimalTmpDirGB) {
tscError("%p sub:%p client disk space remain %.3f GB, need at least %.3f GB, stop query", pPObj, pSql,
tsAvailTmpDirGB, tsMinimalTmpDirGB);
tscAbortFurtherRetryRetrieval(trsupport, tres, TSDB_CODE_CLI_NO_DISKSPACE);
return;
}
// each result for a vnode is ordered as an independant list,
// then used as an input of loser tree for disk-based merge routine
int32_t ret = tscFlushTmpBuffer(trsupport->pExtMemBuffer[idx], pDesc, trsupport->localBuffer,
pQueryInfo->groupbyExpr.orderType);
if (ret != 0) {
/* set no disk space error info, and abort retry */
return tscAbortFurtherRetryRetrieval(trsupport, tres, TSDB_CODE_CLI_NO_DISKSPACE);
}
// keep this value local variable, since the pState variable may be released by other threads, if atomic_add opertion
// increases the finished value up to pState->numOfTotal value, which means all subqueries are completed.
// In this case, the comparsion between finished value and released pState->numOfTotal is not safe.
int32_t numOfTotal = pState->numOfTotal;
int32_t finished = atomic_add_fetch_32(&pState->numOfCompleted, 1);
if (finished < numOfTotal) {
tscTrace("%p sub:%p orderOfSub:%d freed, finished subqueries:%d", pPObj, pSql, trsupport->subqueryIndex, finished);
return tscFreeSubSqlObj(trsupport, pSql);
}
// all sub-queries are returned, start to local merge process
pDesc->pColumnModel->capacity = trsupport->pExtMemBuffer[idx]->numOfElemsPerPage;
tscTrace("%p retrieve from %d vnodes completed.final NumOfRows:%d,start to build loser tree", pPObj,
pState->numOfTotal, pState->numOfRetrievedRows);
SQueryInfo *pPQueryInfo = tscGetQueryInfoDetail(&pPObj->cmd, 0);
tscClearInterpInfo(pPQueryInfo);
tscCreateLocalReducer(trsupport->pExtMemBuffer, pState->numOfTotal, pDesc, trsupport->pFinalColModel,
&pPObj->cmd, &pPObj->res);
tscTrace("%p build loser tree completed", pPObj);
pPObj->res.precision = pSql->res.precision;
pPObj->res.numOfRows = 0;
pPObj->res.row = 0;
// only free once
tfree(trsupport->pState);
tscFreeSubSqlObj(trsupport, pSql);
// set the command flag must be after the semaphore been correctly set.
pPObj->cmd.command = TSDB_SQL_RETRIEVE_METRIC;
if (pPObj->res.code == TSDB_CODE_SUCCESS) {
(*pPObj->fp)(pPObj->param, pPObj, 0);
} else {
tscQueueAsyncRes(pPObj);
taos_fetch_rows_a(tres, tscRetrieveFromDnodeCallBack, param);
}
} else { // all data has been retrieved to client
tscAllDataRetrievedFromDnode(trsupport, pSql);
}
}
......@@ -1409,10 +1422,10 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
SVnodeSidList *vnodeInfo = NULL;
SVnodeDesc * pSvd = NULL;
if (pTableMetaInfo->pMetricMeta != NULL) {
vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
pSvd = &vnodeInfo->vpeerDesc[vnodeInfo->index];
}
// if (pTableMetaInfo->pMetricMeta != NULL) {
// vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
// pSvd = &vnodeInfo->vpeerDesc[vnodeInfo->index];
// }
SSubqueryState* pState = trsupport->pState;
assert(pState->numOfCompleted < pState->numOfTotal && pState->numOfCompleted >= 0 &&
......@@ -1432,7 +1445,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
/*
* if a query on a vnode is failed, all retrieve operations from vnode that occurs later
* than this one are actually not necessary, we simply call the tscRetrieveFromVnodeCallBack
* than this one are actually not necessary, we simply call the tscRetrieveFromDnodeCallBack
* function to abort current and remain retrieve process.
*
* NOTE: threadsafe is required.
......@@ -1453,7 +1466,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY;
} else {
SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
assert(pNewQueryInfo->pTableMetaInfo[0]->pTableMeta != NULL && pNewQueryInfo->pTableMetaInfo[0]->pMetricMeta != NULL);
// assert(pNewQueryInfo->pTableMetaInfo[0]->pTableMeta != NULL && pNewQueryInfo->pTableMetaInfo[0]->pMetricMeta != NULL);
tscProcessSql(pNew);
return;
}
......@@ -1470,7 +1483,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
trsupport->subqueryIndex, pState->code);
}
tscRetrieveFromVnodeCallBack(param, tres, pState->code);
tscRetrieveFromDnodeCallBack(param, tres, pState->code);
} else { // success, proceed to retrieve data from dnode
if (vnodeInfo != NULL) {
tscTrace("%p sub:%p query complete,ip:%u,vid:%d,orderOfSub:%d,retrieve data", trsupport->pParentSqlObj, pSql,
......@@ -1481,7 +1494,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
trsupport->subqueryIndex);
}
taos_fetch_rows_a(tres, tscRetrieveFromVnodeCallBack, param);
taos_fetch_rows_a(tres, tscRetrieveFromDnodeCallBack, param);
}
}
......
src/client/src/tscUtil.c
浏览文件 @ ecae31bb
......@@ -53,7 +53,7 @@ void tscGetMetricMetaCacheKey(SQueryInfo* pQueryInfo, char* str, uint64_t uid) {
const int32_t maxKeySize = TSDB_MAX_TAGS_LEN; // allowed max key size
SCond* cond = tsGetMetricQueryCondPos(pTagCond, uid);
SCond* cond = tsGetSTableQueryCondPos(pTagCond, uid);
char join[512] = {0};
if (pTagCond->joinInfo.hasJoin) {
......@@ -92,7 +92,7 @@ void tscGetMetricMetaCacheKey(SQueryInfo* pQueryInfo, char* str, uint64_t uid) {
free(tmp);
}
SCond* tsGetMetricQueryCondPos(STagCond* pTagCond, uint64_t uid) {
SCond* tsGetSTableQueryCondPos(STagCond* pTagCond, uint64_t uid) {
for (int32_t i = 0; i < TSDB_MAX_JOIN_TABLE_NUM; ++i) {
if (uid == pTagCond->cond[i].uid) {
return &pTagCond->cond[i];
......@@ -102,7 +102,8 @@ SCond* tsGetMetricQueryCondPos(STagCond* pTagCond, uint64_t uid) {
return NULL;
}
void tsSetMetricQueryCond(STagCond* pTagCond, uint64_t uid, const char* str) {
// todo refactor by using SArray
void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, const char* str) {
size_t len = strlen(str);
if (len == 0) {
return;
......@@ -122,7 +123,7 @@ bool tscQueryOnMetric(SSqlCmd* pCmd) {
(pCmd->msgType == TSDB_MSG_TYPE_QUERY);
}
bool tscQueryMetricTags(SQueryInfo* pQueryInfo) {
bool tscQueryTags(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
if (tscSqlExprGet(pQueryInfo, i)->functionId != TSDB_FUNC_TAGPRJ) {
return false;
......@@ -172,6 +173,7 @@ void tscGetDBInfoFromMeterId(char* tableId, char* db) {
}
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx) {
#if 0
if (pMetricmeta == NULL) {
tscError("illegal metricmeta");
return 0;
......@@ -189,6 +191,8 @@ SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx
}
return (SVnodeSidList*)(pMetricmeta->list[vnodeIdx] + (char*)pMetricmeta);
#endif
}
STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
......@@ -221,12 +225,12 @@ bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
// for select query super table, the metricmeta can not be null in any cases.
if (pQueryInfo->command == TSDB_SQL_SELECT && UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
assert(pTableMetaInfo->pMetricMeta != NULL);
// assert(pTableMetaInfo->pMetricMeta != NULL);
}
if (pTableMetaInfo->pMetricMeta == NULL) {
return false;
}
// if (pTableMetaInfo->pMetricMeta == NULL) {
// return false;
// }
if ((pQueryInfo->type & TSDB_QUERY_TYPE_FREE_RESOURCE) == TSDB_QUERY_TYPE_FREE_RESOURCE) {
return false;
......@@ -259,12 +263,12 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) {
}
// only query on tag, not a projection query
if (tscQueryMetricTags(pQueryInfo)) {
if (tscQueryTags(pQueryInfo)) {
return false;
}
// for project query, only the following two function is allowed
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
int32_t functionId = tscSqlExprGet(pQueryInfo, i)->functionId;
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG &&
functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) {
......@@ -1925,7 +1929,7 @@ STableMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, ST
}
pTableMetaInfo->pTableMeta = pTableMeta;
pTableMetaInfo->pMetricMeta = pMetricMeta;
// pTableMetaInfo->pMetricMeta = pMetricMeta;
pTableMetaInfo->numOfTags = numOfTags;
if (tags != NULL) {
......@@ -1975,7 +1979,7 @@ void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache)
}
taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pTableMeta), removeFromCache);
taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pMetricMeta), removeFromCache);
// taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pMetricMeta), removeFromCache);
}
void tscResetForNextRetrieve(SSqlRes* pRes) {
......@@ -2114,15 +2118,15 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
STableMetaInfo* pPrevInfo = tscGetTableMetaInfoFromCmd(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex, 0);
STableMeta* pPrevMeterMeta = taosCacheTransfer(tscCacheHandle, (void**)&pPrevInfo->pTableMeta);
SSuperTableMeta* pPrevMetricMeta = taosCacheTransfer(tscCacheHandle, (void**)&pPrevInfo->pMetricMeta);
// SSuperTableMeta* pPrevMetricMeta = taosCacheTransfer(tscCacheHandle, (void**)&pPrevInfo->pMetricMeta);
pFinalInfo = tscAddMeterMetaInfo(pNewQueryInfo, name, pPrevMeterMeta, pPrevMetricMeta, pTableMetaInfo->numOfTags,
pTableMetaInfo->tagColumnIndex);
// pFinalInfo = tscAddMeterMetaInfo(pNewQueryInfo, name, pPrevMeterMeta, pPrevMetricMeta, pTableMetaInfo->numOfTags,
// pTableMetaInfo->tagColumnIndex);
}
assert(pFinalInfo->pTableMeta != NULL && pNewQueryInfo->numOfTables == 1);
if (UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
assert(pFinalInfo->pMetricMeta != NULL);
// assert(pFinalInfo->pMetricMeta != NULL);
}
tscTrace(
......@@ -2222,13 +2226,13 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
if (!UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo) || (pTableMetaInfo->pMetricMeta == NULL)) {
// if (!UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo) || (pTableMetaInfo->pMetricMeta == NULL)) {
return false;
}
// }
int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
return pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) &&
(!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->vnodeIndex < totalVnode - 1);
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
// return pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) &&
// (!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->vnodeIndex < totalVnode - 1);
}
void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
......@@ -2244,7 +2248,8 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
assert(pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && !tscHasReachLimitation(pQueryInfo, pRes));
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
int32_t totalVnode = 0;
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
while (++pTableMetaInfo->vnodeIndex < totalVnode) {
tscTrace("%p current vnode:%d exhausted, try next:%d. total vnode:%d. current numOfRes:%d", pSql,
......
src/client/tests/timeParseTest.cpp 0 → 100644
浏览文件 @ ecae31bb
#include <gtest/gtest.h>
#include <cassert>
#include <iostream>
#include "taos.h"
#include "tstoken.h"
#include "ttime.h"
#include "tutil.h"
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
extern void deltaToUtcInitOnce();
/* test parse time function */
TEST(testCase, parse_time) {
taos_options(TSDB_OPTION_TIMEZONE, "GMT-8");
deltaToUtcInitOnce();
char t1[] = "2018-1-1 1:1:1.952798";
char t13[] = "1970-1-1 0:0:0";
int64_t time = 0, time1 = 0;
taosParseTime(t1, &time, strlen(t1), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 1514739661952);
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, timezone * MILLISECOND_PER_SECOND);
char t2[] = "2018-1-1T1:1:1.952Z";
taosParseTime(t2, &time, strlen(t2), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 1514739661952 + 28800000);
char t3[] = "2018-1-1 1:01:01.952";
taosParseTime(t3, &time, strlen(t3), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 1514739661952);
char t4[] = "2018-1-1 1:01:01.9";
char t5[] = "2018-1-1 1:01:1.900";
char t6[] = "2018-01-01 1:1:1.90";
char t7[] = "2018-01-01 01:01:01.9";
char t8[] = "2018-01-01 01:01:01.9007865";
taosParseTime(t4, &time, strlen(t4), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t5, &time1, strlen(t5), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
taosParseTime(t4, &time, strlen(t4), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t6, &time1, strlen(t6), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
taosParseTime(t4, &time, strlen(t4), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t7, &time1, strlen(t7), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
taosParseTime(t5, &time, strlen(t5), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t8, &time1, strlen(t8), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
char t9[] = "2017-4-3 1:1:2.980";
char t10[] = "2017-4-3T2:1:2.98+9:00";
taosParseTime(t9, &time, strlen(t9), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t10, &time1, strlen(t10), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
char t11[] = "2017-4-3T2:1:2.98+09:00";
taosParseTime(t11, &time, strlen(t11), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t10, &time1, strlen(t10), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
char t12[] = "2017-4-3T2:1:2.98+0900";
taosParseTime(t11, &time, strlen(t11), TSDB_TIME_PRECISION_MILLI);
taosParseTime(t12, &time1, strlen(t12), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, time1);
taos_options(TSDB_OPTION_TIMEZONE, "UTC");
deltaToUtcInitOnce();
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 0);
taos_options(TSDB_OPTION_TIMEZONE, "Asia/Shanghai");
deltaToUtcInitOnce();
char t14[] = "1970-1-1T0:0:0Z";
taosParseTime(t14, &time, strlen(t14), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 0);
char t40[] = "1970-1-1 0:0:0.999999999";
taosParseTime(t40, &time, strlen(t40), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 999 + timezone * MILLISECOND_PER_SECOND);
char t41[] = "1997-1-1 0:0:0.999999999";
taosParseTime(t41, &time, strlen(t41), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 852048000999);
int64_t k = timezone;
char t42[] = "1997-1-1T0:0:0.999999999Z";
taosParseTime(t42, &time, strlen(t42), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 852048000999 - timezone * MILLISECOND_PER_SECOND);
////////////////////////////////////////////////////////////////////
// illegal timestamp format
char t15[] = "2017-12-33 0:0:0";
EXPECT_EQ(taosParseTime(t15, &time, strlen(t15), TSDB_TIME_PRECISION_MILLI), -1);
char t16[] = "2017-12-31 99:0:0";
EXPECT_EQ(taosParseTime(t16, &time, strlen(t16), TSDB_TIME_PRECISION_MILLI), -1);
char t17[] = "2017-12-31T9:0:0";
EXPECT_EQ(taosParseTime(t17, &time, strlen(t17), TSDB_TIME_PRECISION_MILLI), -1);
char t18[] = "2017-12-31T9:0:0.Z";
EXPECT_EQ(taosParseTime(t18, &time, strlen(t18), TSDB_TIME_PRECISION_MILLI), -1);
char t19[] = "2017-12-31 9:0:0.-1";
EXPECT_EQ(taosParseTime(t19, &time, strlen(t19), TSDB_TIME_PRECISION_MILLI), -1);
char t20[] = "2017-12-31 9:0:0.1+12:99";
EXPECT_EQ(taosParseTime(t20, &time, strlen(t20), TSDB_TIME_PRECISION_MILLI), 0);
EXPECT_EQ(time, 1514682000100);
char t21[] = "2017-12-31T9:0:0.1+12:99";
EXPECT_EQ(taosParseTime(t21, &time, strlen(t21), TSDB_TIME_PRECISION_MILLI), -1);
char t22[] = "2017-12-31 9:0:0.1+13:1";
EXPECT_EQ(taosParseTime(t22, &time, strlen(t22), TSDB_TIME_PRECISION_MILLI), 0);
char t23[] = "2017-12-31T9:0:0.1+13:1";
EXPECT_EQ(taosParseTime(t23, &time, strlen(t23), TSDB_TIME_PRECISION_MILLI), 0);
//======================== add some case ============================//
char b1[] = "9999-12-31 23:59:59.999";
taosParseTime(b1, &time, strlen(b1), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 253402271999999);
char b2[] = "2020-01-01 01:01:01.321";
taosParseTime(b2, &time, strlen(b2), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 1577811661321);
taos_options(TSDB_OPTION_TIMEZONE, "America/New_York");
deltaToUtcInitOnce();
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, 18000 * MILLISECOND_PER_SECOND);
taos_options(TSDB_OPTION_TIMEZONE, "Asia/Tokyo");
deltaToUtcInitOnce();
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, -32400 * MILLISECOND_PER_SECOND);
taos_options(TSDB_OPTION_TIMEZONE, "Asia/Shanghai");
deltaToUtcInitOnce();
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(time, -28800 * MILLISECOND_PER_SECOND);
}
src/common/inc/dataformat.h
浏览文件 @ ecae31bb
......@@ -105,11 +105,36 @@ SDataRow tdDataRowDup(SDataRow row);
// ----------------- Data column structure
typedef struct SDataCol {
int64_t len;
char data[];
int8_t type;
int16_t colId;
int bytes;
int len;
int offset;
void * pData;
} SDataCol;
void tdConvertDataRowToCol(SDataCol *cols, STSchema *pSchema, int *iter);
typedef struct {
int maxRowSize;
int maxCols; // max number of columns
int maxPoints; // max number of points
int numOfPoints;
int numOfCols; // Total number of cols
int sversion; // TODO: set sversion
void * buf;
SDataCol cols[];
} SDataCols;
#define keyCol(pCols) (&((pCols)->cols[0])) // Key column
#define dataColsKeyAt(pCols, idx) ((int64_t *)(keyCol(pCols)->pData))[(idx)]
#define dataColsKeyFirst(pCols) dataColsKeyAt(pCols, 0)
#define dataColsKeyLast(pCols) dataColsKeyAt(pCols, (pCols)->numOfPoints - 1)
SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows);
void tdResetDataCols(SDataCols *pCols);
void tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
void tdFreeDataCols(SDataCols *pCols);
void tdAppendDataRowToDataCol(SDataRow row, SDataCols *pCols);
void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop);
#ifdef __cplusplus
}
......
src/common/src/dataformat.c
浏览文件 @ ecae31bb
......@@ -294,14 +294,79 @@ SDataRow tdDataRowDup(SDataRow row) {
return trow;
}
void tdConvertDataRowToCol(SDataCol *cols, STSchema *pSchema, int *iter) {
int row = *iter;
SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows) {
SDataCols *pCols = (SDataCols *)calloc(1, sizeof(SDataCols) + sizeof(SDataCol) * maxCols);
if (pCols == NULL) return NULL;
pCols->maxRowSize = maxRowSize;
pCols->maxCols = maxCols;
pCols->maxPoints = maxRows;
pCols->buf = malloc(maxRowSize * maxRows);
if (pCols->buf == NULL) {
free(pCols);
return NULL;
}
return pCols;
}
void tdInitDataCols(SDataCols *pCols, STSchema *pSchema) {
// assert(schemaNCols(pSchema) <= pCols->numOfCols);
tdResetDataCols(pCols);
pCols->numOfCols = schemaNCols(pSchema);
pCols->cols[0].pData = pCols->buf;
for (int i = 0; i < schemaNCols(pSchema); i++) {
// TODO
if (i > 0) {
pCols->cols[i].pData = (char *)(pCols->cols[i - 1].pData) + schemaColAt(pSchema, i - 1)->bytes * pCols->maxPoints;
}
pCols->cols[i].type = colType(schemaColAt(pSchema, i));
pCols->cols[i].bytes = colBytes(schemaColAt(pSchema, i));
pCols->cols[i].offset = colOffset(schemaColAt(pSchema, i));
pCols->cols[i].colId = colColId(schemaColAt(pSchema, i));
}
*iter = row + 1;
return pCols;
}
void tdFreeDataCols(SDataCols *pCols) {
if (pCols) {
if (pCols->buf) free(pCols->buf);
free(pCols);
}
}
void tdResetDataCols(SDataCols *pCols) {
pCols->numOfPoints = 0;
for (int i = 0; i < pCols->maxCols; i++) {
pCols->cols[i].len = 0;
}
}
void tdAppendDataRowToDataCol(SDataRow row, SDataCols *pCols) {
TSKEY key = dataRowKey(row);
for (int i = 0; i < pCols->numOfCols; i++) {
SDataCol *pCol = pCols->cols + i;
memcpy((void *)((char *)(pCol->pData) + pCol->len), dataRowAt(row, pCol->offset), pCol->bytes);
pCol->len += pCol->bytes;
}
pCols->numOfPoints++;
}
// Pop pointsToPop points from the SDataCols
void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop) {
int pointsLeft = pCols->numOfPoints - pointsToPop;
for (int iCol = 0; iCol < pCols->numOfCols; iCol++) {
SDataCol *p_col = pCols->cols + iCol;
if (p_col->len > 0) {
p_col->len = TYPE_BYTES[p_col->type] * pointsLeft;
if (pointsLeft > 0) {
memmove((void *)(p_col->pData), (void *)((char *)(p_col->pData) + TYPE_BYTES[p_col->type] * pointsToPop), p_col->len);
}
}
}
pCols->numOfPoints = pointsLeft;
}
/**
......
src/dnode/src/dnodeRead.c
浏览文件 @ ecae31bb
......@@ -243,16 +243,6 @@ static void dnodeContinueExecuteQuery(void* pVnode, void* qhandle, SReadMsg *pMs
taos_queue queue = dnodeGetVnodeRworker(pVnode);
taosWriteQitem(queue, TAOS_QTYPE_RPC, pRead);
// SReadMsg readMsg = {
// .rpcMsg = {0},
// .pCont = qhandle,
// .contLen = 0,
// .pRpcContext = pMsg->pRpcContext,
// };
//
// taos_queue queue = dnodeGetVnodeRworker(pVnode);
// taosWriteQitem(queue, TSDB_MSG_TYPE_QUERY, &readMsg);
}
static void dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg) {
......@@ -261,7 +251,7 @@ static void dnodeProcessQueryMsg(void *pVnode, SReadMsg *pMsg) {
SQInfo* pQInfo = NULL;
if (pMsg->contLen != 0) {
void* tsdb = dnodeGetVnodeTsdb(pVnode);
int32_t code = qCreateQueryInfo(tsdb, pQueryTableMsg, NULL, &pQInfo);
int32_t code = qCreateQueryInfo(tsdb, pQueryTableMsg, &pQInfo);
SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
pRsp->code = code;
......
src/inc/taosmsg.h
浏览文件 @ ecae31bb
......@@ -28,83 +28,83 @@ extern "C" {
#include "trpc.h"
// message type
#define TSDB_MSG_TYPE_REG 1
#define TSDB_MSG_TYPE_REG_RSP 2
#define TSDB_MSG_TYPE_SUBMIT 3
#define TSDB_MSG_TYPE_SUBMIT_RSP 4
#define TSDB_MSG_TYPE_QUERY 5
#define TSDB_MSG_TYPE_QUERY_RSP 6
#define TSDB_MSG_TYPE_RETRIEVE 7
#define TSDB_MSG_TYPE_RETRIEVE_RSP 8
#define TSDB_MSG_TYPE_REG 1
#define TSDB_MSG_TYPE_REG_RSP 2
#define TSDB_MSG_TYPE_SUBMIT 3
#define TSDB_MSG_TYPE_SUBMIT_RSP 4
#define TSDB_MSG_TYPE_QUERY 5
#define TSDB_MSG_TYPE_QUERY_RSP 6
#define TSDB_MSG_TYPE_RETRIEVE 7
#define TSDB_MSG_TYPE_RETRIEVE_RSP 8
// message from mnode to dnode
#define TSDB_MSG_TYPE_MD_CREATE_TABLE 9
#define TSDB_MSG_TYPE_MD_CREATE_TABLE 9
#define TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP 10
#define TSDB_MSG_TYPE_MD_DROP_TABLE 11
#define TSDB_MSG_TYPE_MD_DROP_TABLE_RSP 12
#define TSDB_MSG_TYPE_MD_ALTER_TABLE 13
#define TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP 14
#define TSDB_MSG_TYPE_MD_CREATE_VNODE 15
#define TSDB_MSG_TYPE_MD_DROP_TABLE 11
#define TSDB_MSG_TYPE_MD_DROP_TABLE_RSP 12
#define TSDB_MSG_TYPE_MD_ALTER_TABLE 13
#define TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP 14
#define TSDB_MSG_TYPE_MD_CREATE_VNODE 15
#define TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP 16
#define TSDB_MSG_TYPE_MD_DROP_VNODE 17
#define TSDB_MSG_TYPE_MD_DROP_VNODE_RSP 18
#define TSDB_MSG_TYPE_MD_ALTER_VNODE 19
#define TSDB_MSG_TYPE_MD_ALTER_VNODE_RSP 20
#define TSDB_MSG_TYPE_MD_DROP_STABLE 21
#define TSDB_MSG_TYPE_MD_DROP_STABLE_RSP 22
#define TSDB_MSG_TYPE_MD_ALTER_STREAM 23
#define TSDB_MSG_TYPE_MD_DROP_VNODE 17
#define TSDB_MSG_TYPE_MD_DROP_VNODE_RSP 18
#define TSDB_MSG_TYPE_MD_ALTER_VNODE 19
#define TSDB_MSG_TYPE_MD_ALTER_VNODE_RSP 20
#define TSDB_MSG_TYPE_MD_DROP_STABLE 21
#define TSDB_MSG_TYPE_MD_DROP_STABLE_RSP 22
#define TSDB_MSG_TYPE_MD_ALTER_STREAM 23
#define TSDB_MSG_TYPE_MD_ALTER_STREAM_RSP 24
#define TSDB_MSG_TYPE_MD_CONFIG_DNODE 25
#define TSDB_MSG_TYPE_MD_CONFIG_DNODE 25
#define TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP 26
// message from client to mnode
#define TSDB_MSG_TYPE_CM_CONNECT 31
#define TSDB_MSG_TYPE_CM_CONNECT_RSP 32
#define TSDB_MSG_TYPE_CM_CREATE_ACCT 33
#define TSDB_MSG_TYPE_CM_CREATE_ACCT_RSP 34
#define TSDB_MSG_TYPE_CM_ALTER_ACCT 35
#define TSDB_MSG_TYPE_CM_ALTER_ACCT_RSP 36
#define TSDB_MSG_TYPE_CM_DROP_ACCT 37
#define TSDB_MSG_TYPE_CM_DROP_ACCT_RSP 38
#define TSDB_MSG_TYPE_CM_CREATE_USER 39
#define TSDB_MSG_TYPE_CM_CREATE_USER_RSP 40
#define TSDB_MSG_TYPE_CM_ALTER_USER 41
#define TSDB_MSG_TYPE_CM_ALTER_USER_RSP 42
#define TSDB_MSG_TYPE_CM_DROP_USER 43
#define TSDB_MSG_TYPE_CM_DROP_USER_RSP 44
#define TSDB_MSG_TYPE_CM_CREATE_DNODE 45
#define TSDB_MSG_TYPE_CM_CONNECT 31
#define TSDB_MSG_TYPE_CM_CONNECT_RSP 32
#define TSDB_MSG_TYPE_CM_CREATE_ACCT 33
#define TSDB_MSG_TYPE_CM_CREATE_ACCT_RSP 34
#define TSDB_MSG_TYPE_CM_ALTER_ACCT 35
#define TSDB_MSG_TYPE_CM_ALTER_ACCT_RSP 36
#define TSDB_MSG_TYPE_CM_DROP_ACCT 37
#define TSDB_MSG_TYPE_CM_DROP_ACCT_RSP 38
#define TSDB_MSG_TYPE_CM_CREATE_USER 39
#define TSDB_MSG_TYPE_CM_CREATE_USER_RSP 40
#define TSDB_MSG_TYPE_CM_ALTER_USER 41
#define TSDB_MSG_TYPE_CM_ALTER_USER_RSP 42
#define TSDB_MSG_TYPE_CM_DROP_USER 43
#define TSDB_MSG_TYPE_CM_DROP_USER_RSP 44
#define TSDB_MSG_TYPE_CM_CREATE_DNODE 45
#define TSDB_MSG_TYPE_CM_CREATE_DNODE_RSP 46
#define TSDB_MSG_TYPE_CM_DROP_DNODE 47
#define TSDB_MSG_TYPE_CM_DROP_DNODE_RSP 48
#define TSDB_MSG_TYPE_CM_CONFIG_DNODE TSDB_MSG_TYPE_MD_CONFIG_DNODE
#define TSDB_MSG_TYPE_CM_DROP_DNODE 47
#define TSDB_MSG_TYPE_CM_DROP_DNODE_RSP 48
#define TSDB_MSG_TYPE_CM_CONFIG_DNODE TSDB_MSG_TYPE_MD_CONFIG_DNODE
#define TSDB_MSG_TYPE_CM_CONFIG_DNODE_RSP TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP
#define TSDB_MSG_TYPE_CM_CREATE_DB 49
#define TSDB_MSG_TYPE_CM_CREATE_DB_RSP 50
#define TSDB_MSG_TYPE_CM_DROP_DB 51
#define TSDB_MSG_TYPE_CM_DROP_DB_RSP 52
#define TSDB_MSG_TYPE_CM_USE_DB 53
#define TSDB_MSG_TYPE_CM_USE_DB_RSP 54
#define TSDB_MSG_TYPE_CM_ALTER_DB 55
#define TSDB_MSG_TYPE_CM_ALTER_DB_RSP 56
#define TSDB_MSG_TYPE_CM_CREATE_TABLE 57
#define TSDB_MSG_TYPE_CM_CREATE_DB 49
#define TSDB_MSG_TYPE_CM_CREATE_DB_RSP 50
#define TSDB_MSG_TYPE_CM_DROP_DB 51
#define TSDB_MSG_TYPE_CM_DROP_DB_RSP 52
#define TSDB_MSG_TYPE_CM_USE_DB 53
#define TSDB_MSG_TYPE_CM_USE_DB_RSP 54
#define TSDB_MSG_TYPE_CM_ALTER_DB 55
#define TSDB_MSG_TYPE_CM_ALTER_DB_RSP 56
#define TSDB_MSG_TYPE_CM_CREATE_TABLE 57
#define TSDB_MSG_TYPE_CM_CREATE_TABLE_RSP 58
#define TSDB_MSG_TYPE_CM_DROP_TABLE 59
#define TSDB_MSG_TYPE_CM_DROP_TABLE_RSP 60
#define TSDB_MSG_TYPE_CM_ALTER_TABLE 61
#define TSDB_MSG_TYPE_CM_ALTER_TABLE_RSP 62
#define TSDB_MSG_TYPE_CM_TABLE_META 63
#define TSDB_MSG_TYPE_CM_TABLE_META_RSP 64
#define TSDB_MSG_TYPE_CM_STABLE_VGROUP 65
#define TSDB_MSG_TYPE_CM_DROP_TABLE 59
#define TSDB_MSG_TYPE_CM_DROP_TABLE_RSP 60
#define TSDB_MSG_TYPE_CM_ALTER_TABLE 61
#define TSDB_MSG_TYPE_CM_ALTER_TABLE_RSP 62
#define TSDB_MSG_TYPE_CM_TABLE_META 63
#define TSDB_MSG_TYPE_CM_TABLE_META_RSP 64
#define TSDB_MSG_TYPE_CM_STABLE_VGROUP 65
#define TSDB_MSG_TYPE_CM_STABLE_VGROUP_RSP 66
#define TSDB_MSG_TYPE_CM_TABLES_META 67
#define TSDB_MSG_TYPE_CM_TABLES_META_RSP 68
#define TSDB_MSG_TYPE_CM_ALTER_STREAM 69
#define TSDB_MSG_TYPE_CM_TABLES_META 67
#define TSDB_MSG_TYPE_CM_TABLES_META_RSP 68
#define TSDB_MSG_TYPE_CM_ALTER_STREAM 69
#define TSDB_MSG_TYPE_CM_ALTER_STREAM_RSP 70
#define TSDB_MSG_TYPE_CM_SHOW 71
#define TSDB_MSG_TYPE_CM_SHOW_RSP 72
#define TSDB_MSG_TYPE_CM_KILL_QUERY 73
#define TSDB_MSG_TYPE_CM_KILL_QUERY_RSP 74
#define TSDB_MSG_TYPE_CM_KILL_STREAM 75
#define TSDB_MSG_TYPE_CM_SHOW 71
#define TSDB_MSG_TYPE_CM_SHOW_RSP 72
#define TSDB_MSG_TYPE_CM_KILL_QUERY 73
#define TSDB_MSG_TYPE_CM_KILL_QUERY_RSP 74
#define TSDB_MSG_TYPE_CM_KILL_STREAM 75
#define TSDB_MSG_TYPE_CM_KILL_STREAM_RSP 76
#define TSDB_MSG_TYPE_CM_KILL_CONN 77
#define TSDB_MSG_TYPE_CM_KILL_CONN_RSP 78
......@@ -188,15 +188,6 @@ extern char *taosMsg[];
#pragma pack(push, 1)
// typedef struct {
// int32_t vnode;
// int32_t sid;
// int32_t sversion;
// uint64_t uid;
// int16_t numOfRows;
// char payLoad[];
//} SShellSubmitBlock;
typedef struct {
int32_t numOfVnodes;
} SMsgDesc;
......@@ -206,14 +197,6 @@ typedef struct SMsgHead {
int32_t vgId;
} SMsgHead;
// typedef struct {
// SMsgDesc desc;
// SMsgHead header;
// int16_t import;
// int32_t numOfTables; // total number of sid
// char blks[]; // number of data blocks, each table has at least one data block
//} SShellSubmitMsg;
// Submit message for one table
typedef struct SSubmitBlk {
int64_t uid; // table unique id
......@@ -482,14 +465,15 @@ typedef struct {
int64_t intervalOffset; // start offset for interval query
int64_t slidingTime; // value for sliding window
char slidingTimeUnit; // time interval type, for revisement of interval(1d)
int16_t tagLength; // tag length in current query
uint16_t tagCondLen; // tag length in current query
uint16_t nameCondLen; // table name in/like query expression string length
int16_t numOfGroupCols; // num of group by columns
int16_t orderByIdx;
int16_t orderType; // used in group by xx order by xxx
int16_t orderType; // used in group by xx order by xxx
uint64_t groupbyTagIds;
int64_t limit;
int64_t offset;
int16_t queryType; // denote another query process
uint16_t queryType; // denote another query process
int16_t numOfOutputCols; // final output columns numbers
int16_t interpoType; // interpolate type
uint64_t defaultVal; // default value array list
......@@ -630,14 +614,14 @@ typedef struct {
char tableIds[];
} SCMMultiTableInfoMsg;
typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1];
} SCMSuperTableInfoMsg;
typedef struct SCMSTableVgroupMsg {
char tableId[TSDB_TABLE_ID_LEN];
} SCMSTableVgroupMsg;
typedef struct {
int32_t numOfDnodes;
uint32_t dnodeIps[];
} SCMSuperTableInfoRsp;
} SCMSTableVgroupRspMsg;
typedef struct {
int16_t elemLen;
......@@ -683,14 +667,15 @@ typedef struct {
} SSuperTableMeta;
typedef struct STableMetaMsg {
char tableId[TSDB_TABLE_ID_LEN]; // note: This field must be at the front
int32_t contLen;
uint8_t numOfTags;
uint8_t precision;
uint8_t tableType;
int16_t numOfColumns;
int16_t sversion;
int32_t contLen;
char tableId[TSDB_TABLE_ID_LEN]; // table id
char stableId[TSDB_TABLE_ID_LEN]; // stable name if it is created according to super table
uint8_t numOfTags;
uint8_t precision;
uint8_t tableType;
int16_t numOfColumns;
int16_t sversion;
int8_t numOfVpeers;
SVnodeDesc vpeerDesc[TSDB_VNODES_SUPPORT];
int32_t sid;
......
src/mnode/src/mgmtChildTable.c
浏览文件 @ ecae31bb
......@@ -732,6 +732,7 @@ static int32_t mgmtDoGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STab
pMeta->numOfTags = 0;
pMeta->numOfColumns = htons((int16_t)pTable->superTable->numOfColumns);
pMeta->contLen = sizeof(STableMetaMsg) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable->superTable);
strncpy(pMeta->stableId, pTable->superTable->info.tableId, tListLen(pMeta->stableId));
} else {
pMeta->sversion = htons(pTable->sversion);
pMeta->numOfTags = 0;
......
src/mnode/src/mgmtSuperTable.c
浏览文件 @ ecae31bb
......@@ -216,7 +216,7 @@ void* mgmtGetSuperTable(char *tableId) {
}
static void *mgmtGetSuperTableVgroup(SSuperTableObj *pStable) {
SCMSuperTableInfoRsp *rsp = rpcMallocCont(sizeof(SCMSuperTableInfoRsp) + sizeof(uint32_t) * mgmtGetDnodesNum());
SCMSTableVgroupRspMsg *rsp = rpcMallocCont(sizeof(SCMSTableVgroupRspMsg) + sizeof(uint32_t) * mgmtGetDnodesNum());
rsp->numOfDnodes = htonl(1);
rsp->dnodeIps[0] = htonl(inet_addr(tsPrivateIp));
return rsp;
......@@ -628,14 +628,14 @@ void mgmtGetSuperTableMeta(SQueuedMsg *pMsg, SSuperTableObj *pTable) {
}
static void mgmtProcessSuperTableVgroupMsg(SQueuedMsg *pMsg) {
SCMSuperTableInfoMsg *pInfo = pMsg->pCont;
SCMSTableVgroupMsg *pInfo = pMsg->pCont;
STableInfo *pTable = mgmtGetSuperTable(pInfo->tableId);
if (pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
return;
}
SCMSuperTableInfoRsp *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable);
SCMSTableVgroupRspMsg *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable);
if (pRsp != NULL) {
int32_t msgLen = sizeof(SSuperTableObj) + htonl(pRsp->numOfDnodes) * sizeof(int32_t);
SRpcMsg rpcRsp = {0};
......
src/query/CMakeLists.txt
浏览文件 @ ecae31bb
......@@ -12,4 +12,6 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(query ${SRC})
TARGET_LINK_LIBRARIES(query tsdb tutil m rt)
ENDIF ()
\ No newline at end of file
ENDIF ()
ADD_SUBDIRECTORY(tests)
\ No newline at end of file
src/query/inc/qast.h
浏览文件 @ ecae31bb
......@@ -20,6 +20,7 @@
extern "C" {
#endif
#include <tskiplist.h>
#include "os.h"
#include "taosmsg.h"
......@@ -93,8 +94,7 @@ void tSQLBinaryExprToString(tSQLBinaryExpr *pExpr, char *dst, int32_t *len);
void tSQLBinaryExprDestroy(tSQLBinaryExpr **pExprs, void (*fp)(void*));
void tSQLBinaryExprTraverse(tSQLBinaryExpr *pExprs, struct tSkipList *pSkipList, tQueryResultset *result,
SBinaryFilterSupp *param);
void tSQLBinaryExprTraverse(tSQLBinaryExpr *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param);
void tSQLBinaryExprCalcTraverse(tSQLBinaryExpr *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
char *(*cb)(void *, char *, int32_t));
......
src/query/inc/qsqltype.h
浏览文件 @ ecae31bb
......@@ -54,7 +54,7 @@ enum _sql_type {
TSDB_SQL_CONNECT,
TSDB_SQL_USE_DB, // 30
TSDB_SQL_META,
TSDB_SQL_METRIC,
TSDB_SQL_STABLEVGROUP,
TSDB_SQL_MULTI_META,
TSDB_SQL_HB,
......
src/query/inc/qtsbuf.h
浏览文件 @ ecae31bb
......@@ -48,10 +48,10 @@ typedef struct STSElem {
} STSElem;
typedef struct STSCursor {
int32_t vnodeIndex;
int32_t blockIndex;
int32_t tsIndex;
int32_t order;
int32_t vnodeIndex;
int32_t blockIndex;
int32_t tsIndex;
uint32_t order;
} STSCursor;
typedef struct STSBlock {
......
src/query/inc/queryExecutor.h
浏览文件 @ ecae31bb
......@@ -32,12 +32,6 @@ typedef struct SData {
char data[];
} SData;
enum {
// ST_QUERY_KILLED = 0, // query killed
ST_QUERY_PAUSED = 1, // query paused, due to full of the response buffer
ST_QUERY_COMPLETED = 2, // query completed
};
struct SColumnFilterElem;
typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, char* val1, char* val2);
typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);
......@@ -60,18 +54,20 @@ typedef struct SWindowStatus {
} SWindowStatus;
typedef struct SWindowResult {
uint16_t numOfRows;
uint16_t numOfRows; // number of rows of current time window
SPosInfo pos; // Position of current result in disk-based output buffer
SResultInfo* resultInfo; // For each result column, there is a resultInfo
STimeWindow window; // The time window that current result covers.
SWindowStatus status;
SWindowStatus status; // this result status: closed or opened
} SWindowResult;
typedef struct SResultRec {
int64_t total;
int64_t size;
int64_t capacity;
int32_t threshold; // the threshold size, when the number of rows in result buffer, return to client
int64_t total; // total generated result size in rows
int64_t size; // current result set size in rows
int64_t capacity; // capacity of current result output buffer
// result size threshold in rows. If the result buffer is larger than this, pause query and return to client
int32_t threshold;
} SResultRec;
typedef struct SWindowResInfo {
......@@ -99,7 +95,6 @@ typedef struct SSingleColumnFilterInfo {
void* pData;
} SSingleColumnFilterInfo;
/* intermediate pos during multimeter query involves interval */
typedef struct STableQueryInfo {
int64_t lastKey;
STimeWindow win;
......@@ -107,7 +102,7 @@ typedef struct STableQueryInfo {
int16_t queryRangeSet; // denote if the query range is set, only available for interval query
int64_t tag;
STSCursor cur;
int32_t sid; // for retrieve the page id list
int32_t tid; // for retrieve the page id list
SWindowResInfo windowResInfo;
} STableQueryInfo;
......@@ -116,7 +111,6 @@ typedef struct STableDataInfo {
int32_t numOfBlocks;
int32_t start; // start block index
int32_t tableIndex;
void* pMeterObj;
int32_t groupIdx; // group id in table list
STableQueryInfo* pTableQInfo;
} STableDataInfo;
......@@ -172,7 +166,6 @@ typedef struct SQueryRuntimeEnv {
typedef struct SQInfo {
void* signature;
// void* param; // pointer to the RpcReadMsg
TSKEY startTime;
TSKEY elapsedTime;
int32_t pointsInterpo;
......@@ -203,7 +196,7 @@ typedef struct SQInfo {
* @param pQInfo
* @return
*/
int32_t qCreateQueryInfo(void* pVnode, SQueryTableMsg* pQueryTableMsg, void* param, SQInfo** pQInfo);
int32_t qCreateQueryInfo(void* pVnode, SQueryTableMsg* pQueryTableMsg, SQInfo** pQInfo);
/**
* query on single table
......@@ -211,12 +204,6 @@ int32_t qCreateQueryInfo(void* pVnode, SQueryTableMsg* pQueryTableMsg, void* par
*/
void qTableQuery(SQInfo* pQInfo);
/**
* query on super table
* @param pReadMsg
*/
void qSuperTableQuery(void* pReadMsg);
/**
* wait for the query completed, and retrieve final results to client
* @param pQInfo
......
src/query/src/qast.c
浏览文件 @ ecae31bb
......@@ -14,6 +14,9 @@
*/
#include "qast.h"
#include <tarray.h>
#include <tskiplist.h>
#include "../../client/inc/tschemautil.h"
#include "os.h"
#include "qsqlparser.h"
#include "qsyntaxtreefunction.h"
......@@ -105,7 +108,7 @@ static tSQLSyntaxNode *tSQLSyntaxNodeCreate(SSchema *pSchema, int32_t numOfCols,
return NULL;
}
size_t nodeSize = sizeof(tSQLSyntaxNode);
size_t nodeSize = sizeof(tSQLSyntaxNode);
tSQLSyntaxNode *pNode = NULL;
if (pToken->type == TK_ID || pToken->type == TK_TBNAME) {
......@@ -237,9 +240,7 @@ uint8_t isQueryOnPrimaryKey(const char *primaryColumnName, const tSQLSyntaxNode
}
static tSQLSyntaxNode *createSyntaxTree(SSchema *pSchema, int32_t numOfCols, char *str, int32_t *i) {
SSQLToken t0;
t0 = tStrGetToken(str, i, false, 0, NULL);
SSQLToken t0 = tStrGetToken(str, i, false, 0, NULL);
if (t0.n == 0) {
return NULL;
}
......@@ -341,7 +342,8 @@ void tSQLBinaryExprFromString(tSQLBinaryExpr **pExpr, SSchema *pSchema, int32_t
return;
}
int32_t pos = 0;
int32_t pos = 0;
tSQLSyntaxNode *pStxNode = createSyntaxTree(pSchema, numOfCols, src, &pos);
if (pStxNode != NULL) {
assert(pStxNode->nodeType == TSQL_NODE_EXPR);
......@@ -705,126 +707,120 @@ static bool filterItem(tSQLBinaryExpr *pExpr, const void *pItem, SBinaryFilterSu
* @param pSchema tag schemas
* @param fp filter callback function
*/
static UNUSED_FUNC void tSQLBinaryTraverseOnResult(tSQLBinaryExpr *pExpr, tQueryResultset *pResult, SBinaryFilterSupp *param) {
int32_t n = 0;
for (int32_t i = 0; i < pResult->num; ++i) {
void *pItem = pResult->pRes[i];
static void tSQLBinaryTraverseOnResult(tSQLBinaryExpr *pExpr, SArray *pResult, SBinaryFilterSupp *param) {
size_t size = taosArrayGetSize(pResult);
SArray* array = taosArrayInit(size, POINTER_BYTES);
for (int32_t i = 0; i < size; ++i) {
void *pItem = taosArrayGetP(pResult, i);
if (filterItem(pExpr, pItem, param)) {
pResult->pRes[n++] = pResult->pRes[i];
taosArrayPush(array, &pItem);
}
}
pResult->num = n;
taosArrayCopy(pResult, array);
}
//static void tSQLBinaryTraverseOnSkipList(tSQLBinaryExpr *pExpr, tQueryResultset *pResult, tSkipList *pSkipList,
// SBinaryFilterSupp *param) {
// int32_t n = 0;
// SSkipListIterator iter = {0};
//
// int32_t ret = tSkipListIteratorReset(pSkipList, &iter);
// assert(ret == 0);
//
// pResult->pRes = calloc(pSkipList->nSize, POINTER_BYTES);
//
// while (tSkipListIteratorNext(&iter)) {
// tSkipListNode *pNode = tSkipListIteratorGet(&iter);
// if (filterItem(pExpr, pNode, param)) {
// pResult->pRes[n++] = pNode;
// }
// }
//
// pResult->num = n;
//}
static void tSQLBinaryTraverseOnSkipList(tSQLBinaryExpr *pExpr, SArray *pResult, SSkipList *pSkipList,
SBinaryFilterSupp *param) {
SSkipListIterator* iter = tSkipListCreateIter(pSkipList);
while (tSkipListIterNext(iter)) {
SSkipListNode *pNode = tSkipListIterGet(iter);
if (filterItem(pExpr, pNode, param)) {
taosArrayPush(pResult, SL_GET_NODE_DATA(pNode));
}
}
}
// post-root order traverse syntax tree
//void tSQLBinaryExprTraverse(tSQLBinaryExpr *pExpr, tSkipList *pSkipList, tQueryResultset *result,
// SBinaryFilterSupp *param) {
// if (pExpr == NULL) {
// return;
// }
//
// tSQLSyntaxNode *pLeft = pExpr->pLeft;
// tSQLSyntaxNode *pRight = pExpr->pRight;
//
// // recursive traverse left child branch
// if (pLeft->nodeType == TSQL_NODE_EXPR || pRight->nodeType == TSQL_NODE_EXPR) {
// uint8_t weight = pLeft->pExpr->filterOnPrimaryKey + pRight->pExpr->filterOnPrimaryKey;
//
// if (weight == 0 && result->num > 0 && pSkipList == NULL) {
// /**
// * Perform the filter operation based on the initial filter result, which is obtained from filtering from index.
// * Since no index presented, the filter operation is done by scan all elements in the result set.
// *
// * if the query is a high selectivity filter, only small portion of meters are retrieved.
// */
// tSQLBinaryTraverseOnResult(pExpr, result, param);
// } else if (weight == 0) {
// /**
// * apply the hierarchical expression to every node in skiplist for find the qualified nodes
// */
// assert(result->num == 0);
// tSQLBinaryTraverseOnSkipList(pExpr, result, pSkipList, param);
// } else if (weight == 2 || (weight == 1 && pExpr->nSQLBinaryOptr == TSDB_RELATION_OR)) {
// tQueryResultset rLeft = {0};
// tQueryResultset rRight = {0};
//
// tSQLBinaryExprTraverse(pLeft->pExpr, pSkipList, &rLeft, param);
// tSQLBinaryExprTraverse(pRight->pExpr, pSkipList, &rRight, param);
//
// if (pExpr->nSQLBinaryOptr == TSDB_RELATION_AND) { // CROSS
// intersect(&rLeft, &rRight, result);
// } else if (pExpr->nSQLBinaryOptr == TSDB_RELATION_OR) { // or
// merge(&rLeft, &rRight, result);
// } else {
// assert(false);
// }
//
// free(rLeft.pRes);
// free(rRight.pRes);
// } else {
// /*
// * (weight == 1 && pExpr->nSQLBinaryOptr == TSDB_RELATION_AND) is handled here
// *
// * first, we filter results based on the skiplist index, which is the initial filter stage,
// * then, we conduct the secondary filter operation based on the result from the initial filter stage.
// */
// assert(pExpr->nSQLBinaryOptr == TSDB_RELATION_AND);
//
// tSQLBinaryExpr *pFirst = NULL;
// tSQLBinaryExpr *pSecond = NULL;
// if (pLeft->pExpr->filterOnPrimaryKey == 1) {
// pFirst = pLeft->pExpr;
// pSecond = pRight->pExpr;
// } else {
// pFirst = pRight->pExpr;
// pSecond = pLeft->pExpr;
// }
//
// assert(pFirst != pSecond && pFirst != NULL && pSecond != NULL);
//
// // we filter the result based on the skiplist index in the first place
// tSQLBinaryExprTraverse(pFirst, pSkipList, result, param);
//
// /*
// * recursively perform the filter operation based on the initial results,
// * So, we do not set the skiplist index as a parameter
// */
// tSQLBinaryExprTraverse(pSecond, NULL, result, param);
// }
// } else { // column project
// assert(pLeft->nodeType == TSQL_NODE_COL && pRight->nodeType == TSQL_NODE_VALUE);
//
// param->setupInfoFn(pExpr, param->pExtInfo);
// if (pSkipList == NULL) {
// tSQLListTraverseOnResult(pExpr, param->fp, result);
// } else {
void tSQLBinaryExprTraverse(tSQLBinaryExpr *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param) {
if (pExpr == NULL) {
return;
}
tSQLSyntaxNode *pLeft = pExpr->pLeft;
tSQLSyntaxNode *pRight = pExpr->pRight;
// recursive traverse left child branch
if (pLeft->nodeType == TSQL_NODE_EXPR || pRight->nodeType == TSQL_NODE_EXPR) {
uint8_t weight = pLeft->pExpr->filterOnPrimaryKey + pRight->pExpr->filterOnPrimaryKey;
if (weight == 0 && taosArrayGetSize(result) > 0 && pSkipList == NULL) {
/**
* Perform the filter operation based on the initial filter result, which is obtained from filtering from index.
* Since no index presented, the filter operation is done by scan all elements in the result set.
*
* if the query is a high selectivity filter, only small portion of meters are retrieved.
*/
tSQLBinaryTraverseOnResult(pExpr, result, param);
} else if (weight == 0) {
/**
* apply the hierarchical expression to every node in skiplist for find the qualified nodes
*/
assert(taosArrayGetSize(result) == 0);
tSQLBinaryTraverseOnSkipList(pExpr, result, pSkipList, param);
} else if (weight == 2 || (weight == 1 && pExpr->nSQLBinaryOptr == TSDB_RELATION_OR)) {
tQueryResultset rLeft = {0};
tQueryResultset rRight = {0};
tSQLBinaryExprTraverse(pLeft->pExpr, pSkipList, &rLeft, param);
tSQLBinaryExprTraverse(pRight->pExpr, pSkipList, &rRight, param);
if (pExpr->nSQLBinaryOptr == TSDB_RELATION_AND) { // CROSS
intersect(&rLeft, &rRight, result);
} else if (pExpr->nSQLBinaryOptr == TSDB_RELATION_OR) { // or
merge(&rLeft, &rRight, result);
} else {
assert(false);
}
free(rLeft.pRes);
free(rRight.pRes);
} else {
/*
* (weight == 1 && pExpr->nSQLBinaryOptr == TSDB_RELATION_AND) is handled here
*
* first, we filter results based on the skiplist index, which is the initial filter stage,
* then, we conduct the secondary filter operation based on the result from the initial filter stage.
*/
assert(pExpr->nSQLBinaryOptr == TSDB_RELATION_AND);
tSQLBinaryExpr *pFirst = NULL;
tSQLBinaryExpr *pSecond = NULL;
if (pLeft->pExpr->filterOnPrimaryKey == 1) {
pFirst = pLeft->pExpr;
pSecond = pRight->pExpr;
} else {
pFirst = pRight->pExpr;
pSecond = pLeft->pExpr;
}
assert(pFirst != pSecond && pFirst != NULL && pSecond != NULL);
// we filter the result based on the skiplist index in the first place
tSQLBinaryExprTraverse(pFirst, pSkipList, result, param);
/*
* recursively perform the filter operation based on the initial results,
* So, we do not set the skiplist index as a parameter
*/
tSQLBinaryExprTraverse(pSecond, NULL, result, param);
}
} else { // column project
assert(pLeft->nodeType == TSQL_NODE_COL && pRight->nodeType == TSQL_NODE_VALUE);
param->setupInfoFn(pExpr, param->pExtInfo);
if (pSkipList == NULL) {
tSQLListTraverseOnResult(pExpr, param->fp, result);
} else {
// assert(result->num == 0);
//// tSQLDoFilterInitialResult(pSkipList, param->fp, pExpr->info, result);
// }
// }
//}
// tSQLDoFilterInitialResult(pSkipList, param->fp, pExpr->info, result);
}
}
}
void tSQLBinaryExprCalcTraverse(tSQLBinaryExpr *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
char *(*getSourceDataBlock)(void *, char *, int32_t)) {
......
src/query/src/queryExecutor.c
浏览文件 @ ecae31bb
此差异已折叠。
src/query/tests/CMakeLists.txt 0 → 100644
浏览文件 @ ecae31bb
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)
FIND_LIBRARY(LIB_GTEST_STATIC_DIR libgtest.a /usr/lib/ /usr/local/lib)
IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
MESSAGE(STATUS "gTest library found, build unit test")
INCLUDE_DIRECTORIES(${HEADER_GTEST_INCLUDE_DIR})
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(queryTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(queryTest taos query gtest pthread)
ENDIF()
\ No newline at end of file
src/query/tests/histogramTest.cpp 0 → 100644
浏览文件 @ ecae31bb
#include <gtest/gtest.h>
#include <sys/time.h>
#include <cassert>
#include <iostream>
#include "taos.h"
#include "tsdb.h"
#include "tstoken.h"
#include "tutil.h"
#include "qhistogram.h"
/* test validate the names for table/database */
TEST(testCase, histogram_binary_search) {
SHistogramInfo* pHisto = tHistogramCreate(MAX_HISTOGRAM_BIN);
pHisto->numOfEntries = 10;
for (int32_t i = 0; i < 10; ++i) {
pHisto->elems[i].num = 1;
pHisto->elems[i].val = i;
}
int32_t idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 1);
assert(idx == 1);
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 9);
assert(idx == 9);
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 20);
assert(idx == 10);
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, -1);
assert(idx == 0);
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 3.9);
assert(idx == 4);
free(pHisto);
}
TEST(testCase, histogram_add) {
SHistogramInfo* pHisto = NULL;
/**
* use arrayList, elapsed time is:
* before:
* 10,000,000 45sec, bin:1000 (-O0) / 17sec. bin:1000, (-O3)
*
* after:
*
*/
struct timeval systemTime;
gettimeofday(&systemTime, NULL);
int64_t st =
(int64_t)systemTime.tv_sec * 1000L + (uint64_t)systemTime.tv_usec / 1000;
for (int32_t i = 0; i < 10000; ++i) {
tHistogramAdd(&pHisto, i);
// tHistogramPrint(pHisto);
}
//
gettimeofday(&systemTime, NULL);
int64_t et =
(int64_t)systemTime.tv_sec * 1000L + (uint64_t)systemTime.tv_usec / 1000;
printf("total elapsed time: %ld\n", et - st);
printf("elements: %d, slot:%d \n", pHisto->numOfElems, pHisto->numOfEntries);
tHistogramPrint(pHisto);
printf("%ld\n", tHistogramSum(pHisto, 1.5));
printf("%ld\n", tHistogramSum(pHisto, 2));
printf("%ld\n", tHistogramSum(pHisto, 3));
printf("%ld\n", tHistogramSum(pHisto, 4));
printf("%ld\n", tHistogramSum(pHisto, 5));
printf("%ld\n", tHistogramSum(pHisto, 6));
for (int32_t i = 399; i < 400; ++i) {
printf("val:%d, %ld\n", i, tHistogramSum(pHisto, i));
}
double ratio[] = {0 / 100, 20.0 / 100, 88.0 / 100, 100 / 100};
double* res = tHistogramUniform(pHisto, ratio, 4);
for (int32_t i = 0; i < 4; ++i) {
printf("%f\n", res[i]);
}
SHistogramInfo* pHisto1 = NULL;
for (int32_t i = (90000 - 1); i >= 80000; --i) {
tHistogramAdd(&pHisto1, i);
}
tHistogramPrint(pHisto1);
SHistogramInfo* pRes = tHistogramMerge(pHisto1, pHisto, MAX_HISTOGRAM_BIN);
assert(pRes->numOfElems == pHisto->numOfElems + pHisto1->numOfElems);
tHistogramPrint(pRes);
tHistogramDestroy(&pHisto);
tHistogramDestroy(&pHisto1);
tHistogramDestroy(&pRes);
free(res);
}
TEST(testCase, heapsort) {
// int32_t num = 20;
//
// SHeapEntry* pEntry = tHeapCreate(num);
//
// for(int32_t i=0; i<num; ++i) {
// pEntry[i].val = num - 1 - i;
// }
//
// tHeapSort(pEntry, num);
//
// for(int32_t i=0; i<num; ++i) {
// printf("%lf, ", pEntry[i].val);
// }
//
// printf("\n");
//
// free(pEntry);
}
src/query/tests/patternMatchTest.cpp 0 → 100644
浏览文件 @ ecae31bb
#include <gtest/gtest.h>
#include <sys/time.h>
#include <cassert>
#include <iostream>
#include "tsqlfunction.h"
TEST(testCase, patternMatchTest) {
SPatternCompareInfo info = PATTERN_COMPARE_INFO_INITIALIZER;
const char* str = "abcdef";
int32_t ret = patternMatch("a%b%", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "tm01";
ret = patternMatch("tm__", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "tkm1";
ret = patternMatch("t%m1", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "tkm1";
ret = patternMatch("%m1", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "";
ret = patternMatch("%_", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
str = "1";
ret = patternMatch("%__", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
str = "";
ret = patternMatch("%", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = " ";
ret = patternMatch("_", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "!";
ret = patternMatch("%_", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "abcdefg";
ret = patternMatch("abc%fg", str, strlen(str), &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "abcdefgabcdeju";
ret = patternMatch("abc%fg", str, 7, &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "abcdefgabcdeju";
ret = patternMatch("abc%f_", str, 6, &info);
EXPECT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
str = "abcdefgabcdeju";
ret = patternMatch("abc%f_", str, 1, &info);
EXPECT_EQ(ret, TSDB_PATTERN_NOMATCH);
str = "abcdefgabcdeju";
ret = patternMatch("ab", str, 2, &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "abcdefgabcdeju";
ret = patternMatch("a%", str, 2, &info);
EXPECT_EQ(ret, TSDB_PATTERN_MATCH);
str = "abcdefgabcdeju";
ret = patternMatch("a__", str, 2, &info);
EXPECT_EQ(ret, TSDB_PATTERN_NOMATCH);
}
src/query/tests/resultBufferTest.cpp 0 → 100644
浏览文件 @ ecae31bb
#include <gtest/gtest.h>
#include <cassert>
#include <iostream>
#include "taos.h"
#include "qresultBuf.h"
#include "tsdb.h"
namespace {
// simple test
void simpleTest() {
SDiskbasedResultBuf* pResultBuf = NULL;
int32_t ret = createDiskbasedResultBuffer(&pResultBuf, 1000, 64);
int32_t pageId = 0;
int32_t groupId = 0;
tFilePage* pBufPage = getNewDataBuf(pResultBuf, groupId, &pageId);
ASSERT_TRUE(pBufPage != NULL);
ASSERT_EQ(getNumOfRowsPerPage(pResultBuf), (16384L - sizeof(int64_t))/64);
ASSERT_EQ(getResBufSize(pResultBuf), 1000*16384L);
SIDList list = getDataBufPagesIdList(pResultBuf, groupId);
ASSERT_EQ(list.size, 1);
ASSERT_EQ(getNumOfResultBufGroupId(pResultBuf), 1);
destroyResultBuf(pResultBuf);
}
} // namespace
TEST(testCase, resultBufferTest) {
simpleTest();
}
src/query/tests/tsBufTest.cpp 0 → 100644
浏览文件 @ ecae31bb
#include <gtest/gtest.h>
#include <cassert>
#include <iostream>
#include "taos.h"
#include "tsdb.h"
#include "tstoken.h"
#include "ttime.h"
#include "tutil.h"
#include "qtsbuf.h"
namespace {
/**
*
* @param num total number
* @param step gap between two consecutive ts
* @return
*/
int64_t* createTsList(int32_t num, int64_t start, int32_t step) {
int64_t* pList = (int64_t*)malloc(num * sizeof(int64_t));
for (int64_t i = 0; i < num; ++i) {
pList[i] = start + i * step;
}
return pList;
}
// simple test
void simpleTest() {
STSBuf* pTSBuf = tsBufCreate(true);
// write 10 ts points
int32_t num = 10;
int64_t tag = 1;
int64_t* list = createTsList(10, 10000000, 30);
tsBufAppend(pTSBuf, 0, tag, (const char*)list, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->tsOrder, TSQL_SO_ASC);
EXPECT_EQ(pTSBuf->tsData.len, sizeof(int64_t) * num);
EXPECT_EQ(pTSBuf->block.tag, tag);
EXPECT_EQ(pTSBuf->numOfVnodes, 1);
tsBufFlush(pTSBuf);
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.numOfElem, num);
tsBufDestory(pTSBuf);
}
// one large list of ts, the ts list need to be split into several small blocks
void largeTSTest() {
STSBuf* pTSBuf = tsBufCreate(true);
// write 10 ts points
int32_t num = 1000000;
int64_t tag = 1;
int64_t* list = createTsList(num, 10000000, 30);
tsBufAppend(pTSBuf, 0, tag, (const char*)list, num * sizeof(int64_t));
// the data has been flush to disk, no data in cache
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.tag, tag);
EXPECT_EQ(pTSBuf->numOfVnodes, 1);
EXPECT_EQ(pTSBuf->tsOrder, TSQL_SO_ASC);
tsBufFlush(pTSBuf);
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.numOfElem, num);
tsBufDestory(pTSBuf);
}
void multiTagsTest() {
STSBuf* pTSBuf = tsBufCreate(true);
int32_t num = 10000;
int64_t tag = 1;
int64_t start = 10000000;
int32_t numOfTags = 50;
int32_t step = 30;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf, 0, i, (const char*)list, num * sizeof(int64_t));
free(list);
start += step * num;
}
EXPECT_EQ(pTSBuf->tsOrder, TSQL_SO_ASC);
EXPECT_EQ(pTSBuf->tsData.len, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->block.tag, numOfTags - 1);
EXPECT_EQ(pTSBuf->numOfVnodes, 1);
tsBufFlush(pTSBuf);
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.numOfElem, num);
tsBufDestory(pTSBuf);
}
void multiVnodeTagsTest() {
STSBuf* pTSBuf = tsBufCreate(true);
int32_t num = 10000;
int64_t start = 10000000;
int32_t numOfTags = 50;
int32_t step = 30;
// 2000 vnodes
for (int32_t j = 0; j < 20; ++j) {
// vnodeId:0
start = 10000000;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf, j, i, (const char*)list, num * sizeof(int64_t));
free(list);
start += step * num;
}
EXPECT_EQ(pTSBuf->numOfVnodes, j + 1);
}
EXPECT_EQ(pTSBuf->tsOrder, TSQL_SO_ASC);
EXPECT_EQ(pTSBuf->tsData.len, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->block.tag, numOfTags - 1);
EXPECT_EQ(pTSBuf->tsData.len, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->block.tag, numOfTags - 1);
tsBufFlush(pTSBuf);
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.numOfElem, num);
tsBufDestory(pTSBuf);
}
void loadDataTest() {
STSBuf* pTSBuf = tsBufCreate(true);
int32_t num = 10000;
int64_t oldStart = 10000000;
int32_t numOfTags = 50;
int32_t step = 30;
int32_t numOfVnode = 200;
// 10000 vnodes
for (int32_t j = 0; j < numOfVnode; ++j) {
// vnodeId:0
int64_t start = 10000000;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf, j, i, (const char*)list, num * sizeof(int64_t));
printf("%d - %lld\n", i, list[0]);
free(list);
start += step * num;
}
EXPECT_EQ(pTSBuf->numOfVnodes, j + 1);
}
EXPECT_EQ(pTSBuf->tsOrder, TSQL_SO_ASC);
EXPECT_EQ(pTSBuf->tsData.len, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->block.tag, numOfTags - 1);
EXPECT_EQ(pTSBuf->tsData.len, num * sizeof(int64_t));
EXPECT_EQ(pTSBuf->block.tag, numOfTags - 1);
tsBufFlush(pTSBuf);
EXPECT_EQ(pTSBuf->tsData.len, 0);
EXPECT_EQ(pTSBuf->block.numOfElem, num);
// create from exists file
STSBuf* pNewBuf = tsBufCreateFromFile(pTSBuf->path, false);
EXPECT_EQ(pNewBuf->tsOrder, pTSBuf->tsOrder);
EXPECT_EQ(pNewBuf->numOfVnodes, numOfVnode);
EXPECT_EQ(pNewBuf->fileSize, pTSBuf->fileSize);
EXPECT_EQ(pNewBuf->pData[0].info.offset, pTSBuf->pData[0].info.offset);
EXPECT_EQ(pNewBuf->pData[0].info.numOfBlocks, pTSBuf->pData[0].info.numOfBlocks);
EXPECT_EQ(pNewBuf->pData[0].info.compLen, pTSBuf->pData[0].info.compLen);
EXPECT_STREQ(pNewBuf->path, pTSBuf->path);
tsBufResetPos(pNewBuf);
int64_t s = taosGetTimestampUs();
printf("start:%lld\n", s);
int32_t x = 0;
while (tsBufNextPos(pNewBuf)) {
STSElem elem = tsBufGetElem(pNewBuf);
if (++x == 100000000) {
break;
}
// printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
}
int64_t e = taosGetTimestampUs();
printf("end:%lld, elapsed:%lld, total obj:%d\n", e, e - s, x);
}
void randomIncTsTest() {}
void TSTraverse() {
// 10000 vnodes
int32_t num = 200000;
int64_t oldStart = 10000000;
int32_t numOfTags = 3;
int32_t step = 30;
int32_t numOfVnode = 2;
STSBuf* pTSBuf = tsBufCreate(true);
for (int32_t j = 0; j < numOfVnode; ++j) {
// vnodeId:0
int64_t start = 10000000;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf, j, i, (const char*)list, num * sizeof(int64_t));
printf("%d - %d - %lld, %lld\n", j, i, list[0], list[num - 1]);
free(list);
start += step * num;
list = createTsList(num, start, step);
tsBufAppend(pTSBuf, j, i, (const char*)list, num * sizeof(int64_t));
printf("%d - %d - %lld, %lld\n", j, i, list[0], list[num - 1]);
free(list);
start += step * num;
}
EXPECT_EQ(pTSBuf->numOfVnodes, j + 1);
}
tsBufResetPos(pTSBuf);
////////////////////////////////////////////////////////////////////////////////////////
// reverse traverse
int64_t s = taosGetTimestampUs();
printf("start:%lld\n", s);
pTSBuf->cur.order = TSQL_SO_DESC;
// complete reverse traverse
int32_t x = 0;
while (tsBufNextPos(pTSBuf)) {
STSElem elem = tsBufGetElem(pTSBuf);
// printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
}
// specify the data block with vnode and tags value
tsBufResetPos(pTSBuf);
pTSBuf->cur.order = TSQL_SO_DESC;
int32_t startVnode = 1;
int32_t startTag = 2;
tsBufGetElemStartPos(pTSBuf, startVnode, startTag);
int32_t totalOutput = 10;
while (1) {
STSElem elem = tsBufGetElem(pTSBuf);
printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
if (!tsBufNextPos(pTSBuf)) {
break;
}
if (--totalOutput <= 0) {
totalOutput = 10;
tsBufGetElemStartPos(pTSBuf, startVnode, --startTag);
if (startTag == 0) {
startVnode -= 1;
startTag = 3;
}
if (startVnode < 0) {
break;
}
}
}
/////////////////////////////////////////////////////////////////////////////////
// traverse
pTSBuf->cur.order = TSQL_SO_ASC;
tsBufResetPos(pTSBuf);
// complete forwards traverse
while (tsBufNextPos(pTSBuf)) {
STSElem elem = tsBufGetElem(pTSBuf);
// printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
}
// specify the data block with vnode and tags value
tsBufResetPos(pTSBuf);
pTSBuf->cur.order = TSQL_SO_ASC;
startVnode = 1;
startTag = 2;
tsBufGetElemStartPos(pTSBuf, startVnode, startTag);
totalOutput = 10;
while (1) {
STSElem elem = tsBufGetElem(pTSBuf);
printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
if (!tsBufNextPos(pTSBuf)) {
break;
}
if (--totalOutput <= 0) {
totalOutput = 10;
tsBufGetElemStartPos(pTSBuf, startVnode, --startTag);
if (startTag < 0) {
startVnode -= 1;
startTag = 3;
}
if (startVnode < 0) {
break;
}
}
}
}
void performanceTest() {}
void emptyTagTest() {}
void invalidFileTest() {
const char* cmd = "touch /tmp/test";
// create empty file
system(cmd);
STSBuf* pNewBuf = tsBufCreateFromFile("/tmp/test", true);
EXPECT_TRUE(pNewBuf == NULL);
pNewBuf = tsBufCreateFromFile("/tmp/911", true);
EXPECT_TRUE(pNewBuf == NULL);
}
void mergeDiffVnodeBufferTest() {
STSBuf* pTSBuf1 = tsBufCreate(true);
STSBuf* pTSBuf2 = tsBufCreate(true);
int32_t step = 30;
int32_t num = 1000;
int32_t numOfTags = 10;
// vnodeId:0
int64_t start = 10000000;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf1, 0, i, (const char*)list, num * sizeof(int64_t));
tsBufAppend(pTSBuf2, 0, i, (const char*)list, num * sizeof(int64_t));
free(list);
start += step * num;
}
tsBufFlush(pTSBuf2);
tsBufMerge(pTSBuf1, pTSBuf2, 9);
EXPECT_EQ(pTSBuf1->numOfVnodes, 2);
EXPECT_EQ(pTSBuf1->numOfTotal, numOfTags * 2 * num);
tsBufDisplay(pTSBuf1);
tsBufDestory(pTSBuf2);
tsBufDestory(pTSBuf1);
}
void mergeIdenticalVnodeBufferTest() {
STSBuf* pTSBuf1 = tsBufCreate(true);
STSBuf* pTSBuf2 = tsBufCreate(true);
int32_t step = 30;
int32_t num = 1000;
int32_t numOfTags = 10;
// vnodeId:0
int64_t start = 10000000;
for (int32_t i = 0; i < numOfTags; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf1, 12, i, (const char*)list, num * sizeof(int64_t));
free(list);
start += step * num;
}
for (int32_t i = numOfTags; i < numOfTags * 2; ++i) {
int64_t* list = createTsList(num, start, step);
tsBufAppend(pTSBuf2, 77, i, (const char*)list, num * sizeof(int64_t));
free(list);
start += step * num;
}
tsBufFlush(pTSBuf2);
tsBufMerge(pTSBuf1, pTSBuf2, 12);
EXPECT_EQ(pTSBuf1->numOfVnodes, 1);
EXPECT_EQ(pTSBuf1->numOfTotal, numOfTags * 2 * num);
tsBufResetPos(pTSBuf1);
while (tsBufNextPos(pTSBuf1)) {
STSElem elem = tsBufGetElem(pTSBuf1);
EXPECT_EQ(elem.vnode, 12);
printf("%d-%lld-%lld\n", elem.vnode, elem.tag, elem.ts);
}
tsBufDestory(pTSBuf1);
tsBufDestory(pTSBuf2);
}
} // namespace
TEST(testCase, tsBufTest) {
simpleTest();
largeTSTest();
multiTagsTest();
multiVnodeTagsTest();
loadDataTest();
invalidFileTest();
// randomIncTsTest();
TSTraverse();
mergeDiffVnodeBufferTest();
mergeIdenticalVnodeBufferTest();
}
src/query/tests/unitTest.cpp 0 → 100644
浏览文件 @ ecae31bb
此差异已折叠。
src/util/CMakeLists.txt
浏览文件 @ ecae31bb
......@@ -24,6 +24,8 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
ELSE ()
MESSAGE(STATUS "Failed to find iconv, use default encoding method")
ENDIF ()
ADD_SUBDIRECTORY(tests)
ELSEIF (TD_WINDOWS_64)
ADD_DEFINITIONS(-DUSE_LIBICONV)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/pthread)
......@@ -108,4 +110,3 @@ ENDIF()
#ENDIF ()
src/util/inc/tarray.h
浏览文件 @ ecae31bb
......@@ -87,7 +87,21 @@ size_t taosArrayGetSize(const SArray* pArray);
void* taosArrayInsert(SArray* pArray, size_t index, void* pData);
/**
*
* remove data entry of the given index
* @param pArray
* @param index
*/
void taosArrayRemove(SArray* pArray, size_t index);
/**
* copy the whole array from source to destination
* @param pDst
* @param pSrc
*/
void taosArrayCopy(SArray* pDst, SArray* pSrc);
/**
* destroy array list
* @param pArray
*/
void taosArrayDestroy(SArray* pArray);
......
src/util/inc/tskiplist.h
浏览文件 @ ecae31bb
此差异已折叠。
src/util/inc/tutil.h
浏览文件 @ ecae31bb
......@@ -176,6 +176,14 @@ uint32_t ip2uint(const char *const ip_addr);
void taosSetAllocMode(int mode, const char* path, bool autoDump);
void taosDumpMemoryLeak();
#define TD_EQ 0x1
#define TD_GT 0x2
#define TD_LT 0x4
#define TD_GE (TD_EQ | TD_GT)
#define TD_LE (TD_EQ | TD_LT)
void *taosbsearch(const void *key, const void *base, size_t nmemb, size_t size,
int (*compar)(const void *, const void *), int flags);
#ifdef TAOS_MEM_CHECK
void * taos_malloc(size_t size, const char *file, uint32_t line);
......
src/util/src/tarray.c
浏览文件 @ ecae31bb
......@@ -128,6 +128,38 @@ void* taosArrayInsert(SArray* pArray, size_t index, void* pData) {
return dst;
}
void taosArrayRemove(SArray* pArray, size_t index) {
assert(index < pArray->size);
if (index == pArray->size - 1) {
taosArrayPop(pArray);
return;
}
size_t remain = pArray->size - index - 1;
memmove(pArray->pData + index * pArray->elemSize, pArray->pData + (index + 1) * pArray->elemSize, remain * pArray->elemSize);
pArray->size -= 1;
}
void taosArrayCopy(SArray* pDst, SArray* pSrc) {
assert(pSrc != NULL && pDst != NULL);
if (pDst->capacity < pSrc->size) {
void* pData = realloc(pDst->pData, pSrc->size * pSrc->elemSize);
if (pData == NULL) { // todo handle oom
} else {
pDst->pData = pData;
pDst->capacity = pSrc->size;
}
}
memcpy(pDst->pData, pSrc->pData, pSrc->elemSize * pSrc->size);
pDst->elemSize = pSrc->elemSize;
pDst->capacity = pSrc->size;
pDst->size = pSrc->size;
}
void taosArrayDestroy(SArray* pArray) {
if (pArray == NULL) {
return;
......
src/util/src/tlist.c
浏览文件 @ ecae31bb
此差异已折叠。
src/util/src/tskiplist.c
浏览文件 @ ecae31bb
此差异已折叠。
src/util/src/tutil.c
浏览文件 @ ecae31bb
此差异已折叠。
src/util/tests/CMakeLists.txt 0 → 100644
浏览文件 @ ecae31bb
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)
FIND_LIBRARY(LIB_GTEST_STATIC_DIR libgtest.a /usr/lib/ /usr/local/lib)
IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
MESSAGE(STATUS "gTest library found, build unit test")
INCLUDE_DIRECTORIES(${HEADER_GTEST_INCLUDE_DIR})
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(utilTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(utilTest tutil gtest pthread)
ENDIF()
\ No newline at end of file
src/util/tests/cacheTest.cpp 0 → 100644
浏览文件 @ ecae31bb
此差异已折叠。
src/util/tests/hashTest.cpp 0 → 100644
浏览文件 @ ecae31bb
此差异已折叠。
src/util/tests/skiplistTest.cpp 0 → 100644
浏览文件 @ ecae31bb
此差异已折叠。
src/util/tests/stringTest.cpp 0 → 100644
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/inc/tsdb.h
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/inc/tsdbFile.h
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/src/tsdbCache.c
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/src/tsdbFile.c
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/src/tsdbMain.c
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/src/tsdbMeta.c
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/src/tsdbRead.c
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/tests/CMakeLists.txt
浏览文件 @ ecae31bb
此差异已折叠。
src/vnode/tsdb/tests/tsdbTests.cpp
浏览文件 @ ecae31bb
此差异已折叠。
tests/examples/c/demo.c
浏览文件 @ ecae31bb
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或