diff --git a/src/client/inc/tscSecondaryMerge.h b/src/client/inc/tscLocalMerge.h similarity index 100% rename from src/client/inc/tscSecondaryMerge.h rename to src/client/inc/tscLocalMerge.h diff --git a/src/client/inc/tscUtil.h b/src/client/inc/tscUtil.h index b3e05cb0ca..581ce00a62 100644 --- a/src/client/inc/tscUtil.h +++ b/src/client/inc/tscUtil.h @@ -28,7 +28,7 @@ extern "C" { #include "exception.h" #include "qextbuffer.h" #include "taosdef.h" -#include "tscSecondaryMerge.h" +#include "tscLocalMerge.h" #include "tsclient.h" #define UTIL_TABLE_IS_SUPER_TABLE(metaInfo) \ @@ -122,15 +122,13 @@ bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo *pQueryInfo, int32_t tableI bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex); bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex); -bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo); +bool tscIsProjectionQuery(SQueryInfo* pQueryInfo); bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex); -bool tscQueryOnSTable(SSqlCmd* pCmd); bool tscQueryTags(SQueryInfo* pQueryInfo); -bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd); void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex, int16_t functionId, SColumnIndex* pIndex, - SSchema* pColSchema, int16_t isTag); + SSchema* pColSchema, int16_t colType); int32_t tscSetTableFullName(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SSqlObj* pSql); void tscClearInterpInfo(SQueryInfo* pQueryInfo); @@ -139,7 +137,7 @@ bool tscIsInsertData(char* sqlstr); /* use for keep current db info temporarily, for handle table with db prefix */ // todo remove it -void tscGetDBInfoFromMeterId(char* tableId, char* db); +void tscGetDBInfoFromTableFullName(char* tableId, char* db); int tscAllocPayload(SSqlCmd* pCmd, int size); @@ -253,7 +251,7 @@ void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t sub void doAddGroupColumnForSubquery(SQueryInfo* pQueryInfo, int32_t tagIndex); -int16_t tscGetJoinTagColIndexByUid(STagCond* pTagCond, uint64_t uid); +int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid); void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex); diff --git a/src/client/inc/tschemautil.h b/src/client/inc/tschemautil.h index 2fc77d69bb..99ffa4e766 100644 --- a/src/client/inc/tschemautil.h +++ b/src/client/inc/tschemautil.h @@ -64,12 +64,20 @@ SSchema* tscGetTableSchema(const STableMeta* pTableMeta); SSchema *tscGetTableTagSchema(const STableMeta *pMeta); /** - * + * get the column schema according to the column index * @param pMeta - * @param startCol + * @param colIndex * @return */ -SSchema *tscGetTableColumnSchema(const STableMeta *pMeta, int32_t startCol); +SSchema *tscGetTableColumnSchema(const STableMeta *pMeta, int32_t colIndex); + +/** + * get the column schema according to the column id + * @param pTableMeta + * @param colId + * @return + */ +SSchema* tscGetTableColumnSchemaById(STableMeta* pTableMeta, int16_t colId); /** * check if the schema is valid or not, including following aspects: diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index ed3b481d0e..e538281bc0 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -85,7 +85,7 @@ typedef struct SSqlExpr { int16_t functionId; // function id in aAgg array int16_t resType; // return value type int16_t resBytes; // length of return value - int32_t interBytes; // inter result buffer size + int32_t interBytes; // inter result buffer size int16_t numOfParams; // argument value of each function tVariant param[3]; // parameters are not more than 3 int32_t offset; // sub result column value of arithmetic expression. @@ -123,7 +123,7 @@ typedef struct SCond { typedef struct SJoinNode { char tableId[TSDB_TABLE_ID_LEN]; uint64_t uid; - int16_t tagCol; + int16_t tagColId; } SJoinNode; typedef struct SJoinInfo { @@ -155,20 +155,19 @@ typedef struct SParamInfo { } SParamInfo; typedef struct STableDataBlocks { - char tableId[TSDB_TABLE_ID_LEN]; - int8_t tsSource; // where does the UNIX timestamp come from, server or client - bool ordered; // if current rows are ordered or not - int64_t vgId; // virtual group id - int64_t prevTS; // previous timestamp, recorded to decide if the records array is ts ascending - int32_t numOfTables; // number of tables in current submit block - - int32_t rowSize; // row size for current table + char tableId[TSDB_TABLE_ID_LEN]; + int8_t tsSource; // where does the UNIX timestamp come from, server or client + bool ordered; // if current rows are ordered or not + int64_t vgId; // virtual group id + int64_t prevTS; // previous timestamp, recorded to decide if the records array is ts ascending + int32_t numOfTables; // number of tables in current submit block + int32_t rowSize; // row size for current table uint32_t nAllocSize; - uint32_t headerSize; // header for metadata (submit metadata) + uint32_t headerSize; // header for table info (uid, tid, submit metadata) uint32_t size; /* - * the metermeta for current table, the metermeta will be used during submit stage, keep a ref + * the table meta of table, the table meta will be used during submit, keep a ref * to avoid it to be removed from cache */ STableMeta *pTableMeta; @@ -191,32 +190,28 @@ typedef struct SDataBlockList { // todo remove } SDataBlockList; typedef struct SQueryInfo { - int16_t command; // the command may be different for each subclause, so keep it seperately. - uint32_t type; // query/insert/import type + int16_t command; // the command may be different for each subclause, so keep it seperately. + uint32_t type; // query/insert/import type char slidingTimeUnit; - STimeWindow window; int64_t intervalTime; // aggregation time interval int64_t slidingTime; // sliding window in mseconds SSqlGroupbyExpr groupbyExpr; // group by tags info - - SArray * colList; // SArray + SArray * colList; // SArray SFieldInfo fieldsInfo; - SArray * exprList; // SArray + SArray * exprList; // SArray SLimitVal limit; SLimitVal slimit; STagCond tagCond; SOrderVal order; - int16_t fillType; // final result fill type + int16_t fillType; // final result fill type int16_t numOfTables; STableMetaInfo **pTableMetaInfo; struct STSBuf * tsBuf; - int64_t * fillVal; // default value for fill - char * msg; // pointer to the pCmd->payload to keep error message temporarily - int64_t clauseLimit; // limit for current sub clause - - // offset value in the original sql expression, NOT sent to virtual node, only applied at client side - int64_t prjOffset; + int64_t * fillVal; // default value for fill + char * msg; // pointer to the pCmd->payload to keep error message temporarily + int64_t clauseLimit; // limit for current sub clause + int64_t prjOffset; // offset value in the original sql expression, only applied at client side } SQueryInfo; typedef struct { @@ -431,7 +426,7 @@ extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo); typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows); int32_t tscCompareTidTags(const void* p1, const void* p2); -void tscBuildVgroupTableInfo(STableMetaInfo* pTableMetaInfo, SArray* tables); +void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables); #ifdef __cplusplus } diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index 770f328634..3fedc9e7ad 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -19,9 +19,8 @@ #include "tnote.h" #include "trpc.h" #include "tscLog.h" -#include "tscProfile.h" #include "tscSubquery.h" -#include "tscSecondaryMerge.h" +#include "tscLocalMerge.h" #include "tscUtil.h" #include "tsched.h" #include "tschemautil.h" @@ -46,7 +45,8 @@ int doAsyncParseSql(SSqlObj* pSql) { int32_t code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE); if (code != TSDB_CODE_SUCCESS) { tscError("failed to malloc payload"); - tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY); + tscQueueAsyncRes(pSql); +// tscQueueAsyncRes(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY); return code; } @@ -211,7 +211,8 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi if (pRes->qhandle == 0) { tscError("qhandle is NULL"); - tscQueueAsyncError(fp, param, TSDB_CODE_TSC_INVALID_QHANDLE); + pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE; + tscQueueAsyncRes(pSql); return; } diff --git a/src/client/src/tscFunctionImpl.c b/src/client/src/tscFunctionImpl.c index f0be65882c..d5e3126f71 100644 --- a/src/client/src/tscFunctionImpl.c +++ b/src/client/src/tscFunctionImpl.c @@ -153,7 +153,7 @@ typedef struct SRateInfo { int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, int16_t *type, int16_t *bytes, int32_t *interBytes, int16_t extLength, bool isSuperTable) { - if (!isValidDataType(dataType, dataBytes)) { + if (!isValidDataType(dataType)) { tscError("Illegal data type %d or data type length %d", dataType, dataBytes); return TSDB_CODE_TSC_INVALID_SQL; } @@ -2977,12 +2977,12 @@ static void tag_project_function_f(SQLFunctionCtx *pCtx, int32_t index) { */ static void tag_function(SQLFunctionCtx *pCtx) { SET_VAL(pCtx, 1, 1); - tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true); + tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->outputType, true); } static void tag_function_f(SQLFunctionCtx *pCtx, int32_t index) { SET_VAL(pCtx, 1, 1); - tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true); + tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->outputType, true); } static void copy_function(SQLFunctionCtx *pCtx) { @@ -3891,7 +3891,7 @@ static bool ts_comp_function_setup(SQLFunctionCtx *pCtx) { SResultInfo *pResInfo = GET_RES_INFO(pCtx); STSCompInfo *pInfo = pResInfo->interResultBuf; - pInfo->pTSBuf = tsBufCreate(false); + pInfo->pTSBuf = tsBufCreate(false, pCtx->order); pInfo->pTSBuf->tsOrder = pCtx->order; return true; } @@ -3913,7 +3913,6 @@ static void ts_comp_function(SQLFunctionCtx *pCtx) { } SET_VAL(pCtx, pCtx->size, 1); - pResInfo->hasResult = DATA_SET_FLAG; } diff --git a/src/client/src/tscLocal.c b/src/client/src/tscLocal.c index b1a6e1595e..83700ce0a5 100644 --- a/src/client/src/tscLocal.c +++ b/src/client/src/tscLocal.c @@ -22,7 +22,6 @@ #include "taosdef.h" #include "tscLog.h" #include "qextbuffer.h" -#include "tscSecondaryMerge.h" #include "tschemautil.h" #include "tname.h" diff --git a/src/client/src/tscSecondaryMerge.c b/src/client/src/tscLocalMerge.c similarity index 99% rename from src/client/src/tscSecondaryMerge.c rename to src/client/src/tscLocalMerge.c index 50cda07ae0..f64287bdc4 100644 --- a/src/client/src/tscSecondaryMerge.c +++ b/src/client/src/tscLocalMerge.c @@ -13,7 +13,6 @@ * along with this program. If not, see . */ -#include "tscSecondaryMerge.h" #include "os.h" #include "tlosertree.h" #include "tscUtil.h" @@ -21,6 +20,7 @@ #include "tsclient.h" #include "tutil.h" #include "tscLog.h" +#include "tscLocalMerge.h" typedef struct SCompareParam { SLocalDataSource **pLocalData; @@ -1096,14 +1096,6 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx) size_t size = tscSqlExprNumOfExprs(pQueryInfo); for (int32_t j = 0; j < size; ++j) { - // SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[j]; - // if (pExpr == NULL) { - // assert(pQueryInfo->fieldsInfo.pExpr[j] != NULL); - // - // maxOutput = 1; - // continue; - // } - /* * ts, tag, tagprj function can not decide the output number of current query * the number of output result is decided by main output @@ -1113,8 +1105,9 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx) continue; } - if (maxOutput < GET_RES_INFO(&pCtx[j])->numOfRes) { - maxOutput = GET_RES_INFO(&pCtx[j])->numOfRes; + SResultInfo* pResInfo = GET_RES_INFO(&pCtx[j]); + if (maxOutput < pResInfo->numOfRes) { + maxOutput = pResInfo->numOfRes; } } @@ -1264,7 +1257,6 @@ bool doGenerateFinalResults(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool no #ifdef _DEBUG_VIEW printf("final result before interpo:\n"); - assert(0); // tColModelDisplay(pLocalReducer->resColModel, pLocalReducer->pBufForInterpo, pResBuf->num, pResBuf->num); #endif diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index 13fc298d6e..d821271857 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -1327,12 +1327,14 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) { int32_t ret = TSDB_CODE_SUCCESS; if (initialParse) { + assert(!pSql->cmd.parseFinished); + char* p = pSql->sqlstr; pSql->sqlstr = NULL; tscPartiallyFreeSqlObj(pSql); pSql->sqlstr = p; - } else { + } else if (!pSql->cmd.parseFinished) { tscTrace("continue parse sql: %s", pSql->cmd.curSql); } diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 110137ba5e..494540290b 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -98,8 +98,6 @@ static int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo, int32_t killTy static bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField); static bool hasTimestampForPointInterpQuery(SQueryInfo* pQueryInfo); -static void updateTagColumnIndex(SQueryInfo* pQueryInfo, int32_t tableIndex); - static int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t index, SQuerySQL* pQuerySql, SSqlObj* pSql); static int32_t parseCreateDBOptions(SSqlCmd* pCmd, SCreateDBInfo* pCreateDbSql); static int32_t getColumnIndexByName(const SSQLToken* pToken, SQueryInfo* pQueryInfo, SColumnIndex* pIndex); @@ -640,17 +638,11 @@ int32_t parseIntervalClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) { return TSDB_CODE_TSC_INVALID_SQL; } + SSchema s = {.bytes = TSDB_KEYSIZE, .type = TSDB_DATA_TYPE_TIMESTAMP, .colId = PRIMARYKEY_TIMESTAMP_COL_INDEX}; + tstrncpy(s.name, aAggs[TSDB_FUNC_TS].aName, sizeof(s.name)); + SColumnIndex index = {tableIndex, PRIMARYKEY_TIMESTAMP_COL_INDEX}; - SSqlExpr* pExpr = tscSqlExprInsert(pQueryInfo, 0, TSDB_FUNC_TS, &index, TSDB_DATA_TYPE_TIMESTAMP, TSDB_KEYSIZE, - TSDB_KEYSIZE, false); - - SColumnList ids = getColumnList(1, 0, PRIMARYKEY_TIMESTAMP_COL_INDEX); - - int32_t ret = - insertResultField(pQueryInfo, 0, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, aAggs[TSDB_FUNC_TS].aName, pExpr); - if (ret != TSDB_CODE_SUCCESS) { - return ret; - } + tscAddSpecialColumnForSelect(pQueryInfo, 0, TSDB_FUNC_TS, &index, &s, TSDB_COL_NORMAL); if (parseSlidingClause(pQueryInfo, pQuerySql) != TSDB_CODE_SUCCESS) { return TSDB_CODE_TSC_INVALID_SQL; @@ -1241,11 +1233,11 @@ int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, tSQLExprList* pSel return invalidSqlErrMsg(pQueryInfo->msg, msg2); } + /* + * transfer sql functions that need secondary merge into another format + * in dealing with metric queries such as: count/first/last + */ if (isSTable) { - /* - * transfer sql functions that need secondary merge into another format - * in dealing with metric queries such as: count/first/last - */ tscTansformSQLFuncForSTableQuery(pQueryInfo); if (hasUnsupportFunctionsForSTableQuery(pQueryInfo)) { @@ -1272,7 +1264,7 @@ int32_t insertResultField(SQueryInfo* pQueryInfo, int32_t outputIndex, SColumnLi } TAOS_FIELD f = tscCreateField(type, fieldName, bytes); - SFieldSupInfo* pInfo =tscFieldInfoInsert(&pQueryInfo->fieldsInfo, outputIndex, &f); + SFieldSupInfo* pInfo = tscFieldInfoInsert(&pQueryInfo->fieldsInfo, outputIndex, &f); pInfo->pSqlExpr = pSqlExpr; return TSDB_CODE_SUCCESS; @@ -1324,8 +1316,9 @@ static void addProjectQueryCol(SQueryInfo* pQueryInfo, int32_t startPos, SColumn void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex, int16_t functionId, SColumnIndex* pIndex, SSchema* pColSchema, int16_t flag) { - SSqlExpr* pExpr = tscSqlExprAppend(pQueryInfo, functionId, pIndex, pColSchema->type, + SSqlExpr* pExpr = tscSqlExprInsert(pQueryInfo, outputColIndex, functionId, pIndex, pColSchema->type, pColSchema->bytes, pColSchema->bytes, flag); + tstrncpy(pExpr->aliasName, pColSchema->name, sizeof(pExpr->aliasName)); SColumnList ids = getColumnList(1, pIndex->tableIndex, pIndex->columnIndex); if (TSDB_COL_IS_TAG(flag)) { @@ -1403,7 +1396,7 @@ int32_t addProjectionExprAndResultField(SQueryInfo* pQueryInfo, tSQLExprItem* pI if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) { SSchema colSchema = tGetTableNameColumnSchema(); - tscAddSpecialColumnForSelect(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, true); + tscAddSpecialColumnForSelect(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, TSDB_COL_TAG); } else { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; @@ -1846,7 +1839,7 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr } else { tVariantDump(pVariant, val, TSDB_DATA_TYPE_BIGINT, true); - int64_t nTop = *((int32_t*)val); + int64_t nTop = GET_INT32_VAL(val); if (nTop <= 0 || nTop > 100) { // todo use macro return invalidSqlErrMsg(pQueryInfo->msg, msg5); } @@ -1856,12 +1849,14 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr return TSDB_CODE_TSC_INVALID_SQL; } + // todo REFACTOR // set the first column ts for top/bottom query SColumnIndex index1 = {0, PRIMARYKEY_TIMESTAMP_COL_INDEX}; pExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS, &index1, TSDB_DATA_TYPE_TIMESTAMP, TSDB_KEYSIZE, TSDB_KEYSIZE, false); + tstrncpy(pExpr->aliasName, aAggs[TSDB_FUNC_TS].aName, sizeof(pExpr->aliasName)); - const int32_t TS_COLUMN_INDEX = 0; + const int32_t TS_COLUMN_INDEX = PRIMARYKEY_TIMESTAMP_COL_INDEX; SColumnList ids = getColumnList(1, 0, TS_COLUMN_INDEX); insertResultField(pQueryInfo, TS_COLUMN_INDEX, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, aAggs[TSDB_FUNC_TS].aName, pExpr); @@ -2469,62 +2464,10 @@ static bool functionCompatibleCheck(SQueryInfo* pQueryInfo) { return true; } -void updateTagColumnIndex(SQueryInfo* pQueryInfo, int32_t tableIndex) { -// STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex); -// -// // update tags column index for expression -// size_t size = tscSqlExprNumOfExprs(pQueryInfo); -// for (int32_t i = 0; i < size; ++i) { -// SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i); -// -// if (!TSDB_COL_IS_TAG(pExpr->colInfo.flag)) { // not tags, continue -// continue; -// } -// -// // not belongs to this table -// if (pExpr->uid != pTableMetaInfo->pTableMeta->uid) { -// continue; -// } - -// for (int32_t j = 0; j < pTableMetaInfo->numOfTags; ++j) { -// if (pExpr->colInfo.colIndex == pTableMetaInfo->tagColumnIndex[j]) { -// pExpr->colInfo.colIndex = j; -// break; -// } -// } -// } - - // update join condition tag column index -// SJoinInfo* pJoinInfo = &pQueryInfo->tagCond.joinInfo; -// if (!pJoinInfo->hasJoin) { // not join query -// return; -// } -// -// assert(pJoinInfo->left.uid != pJoinInfo->right.uid); -// -// // the join condition expression node belongs to this table(super table) -// assert(0); -// if (pTableMetaInfo->pTableMeta->uid == pJoinInfo->left.uid) { -// for (int32_t i = 0; i < pTableMetaInfo->numOfTags; ++i) { -// if (pJoinInfo->left.tagCol == pTableMetaInfo->tagColumnIndex[i]) { -// pJoinInfo->left.tagCol = i; -// } -// } -// } -// -// if (pTableMetaInfo->pTableMeta->uid == pJoinInfo->right.uid) { -// for (int32_t i = 0; i < pTableMetaInfo->numOfTags; ++i) { -// if (pJoinInfo->right.tagCol == pTableMetaInfo->tagColumnIndex[i]) { -// pJoinInfo->right.tagCol = i; -// } -// } -// } -} - int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd* pCmd) { const char* msg1 = "too many columns in group by clause"; const char* msg2 = "invalid column name in group by clause"; - const char* msg3 = "group by columns must belong to one table"; +// const char* msg3 = "group by columns must belong to one table"; const char* msg7 = "not support group by expression"; const char* msg8 = "not allowed column type for group by"; const char* msg9 = "tags not allowed for table query"; @@ -2560,10 +2503,6 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd* return invalidSqlErrMsg(pQueryInfo->msg, msg2); } - if (tableIndex != index.tableIndex && tableIndex >= 0) { - return invalidSqlErrMsg(pQueryInfo->msg, msg3); - } - tableIndex = index.tableIndex; pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); @@ -2620,7 +2559,6 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd* } pQueryInfo->groupbyExpr.tableIndex = tableIndex; - return TSDB_CODE_SUCCESS; } @@ -3050,14 +2988,17 @@ static int32_t getColumnQueryCondInfo(SQueryInfo* pQueryInfo, tSQLExpr* pExpr, i } static int32_t getJoinCondInfo(SQueryInfo* pQueryInfo, tSQLExpr* pExpr) { - const char* msg = "invalid join query condition"; + const char* msg1 = "invalid join query condition"; + const char* msg2 = "join on binary/nchar not supported"; + const char* msg3 = "type of join columns must be identical"; + const char* msg4 = "invalid column name in join condition"; if (pExpr == NULL) { return TSDB_CODE_SUCCESS; } if (!isExprDirectParentOfLeaftNode(pExpr)) { - return invalidSqlErrMsg(pQueryInfo->msg, msg); + return invalidSqlErrMsg(pQueryInfo->msg, msg1); } STagCond* pTagCond = &pQueryInfo->tagCond; @@ -3066,28 +3007,36 @@ static int32_t getJoinCondInfo(SQueryInfo* pQueryInfo, tSQLExpr* pExpr) { SColumnIndex index = COLUMN_INDEX_INITIALIZER; if (getColumnIndexByName(&pExpr->pLeft->colInfo, pQueryInfo, &index) != TSDB_CODE_SUCCESS) { - return TSDB_CODE_TSC_INVALID_SQL; + return invalidSqlErrMsg(pQueryInfo->msg, msg4); } STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); - int16_t tagColIndex = index.columnIndex - tscGetNumOfColumns(pTableMetaInfo->pTableMeta); + SSchema* pTagSchema1 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index.columnIndex); pLeft->uid = pTableMetaInfo->pTableMeta->uid; - pLeft->tagCol = tagColIndex; + pLeft->tagColId = pTagSchema1->colId; strcpy(pLeft->tableId, pTableMetaInfo->name); index = (SColumnIndex)COLUMN_INDEX_INITIALIZER; if (getColumnIndexByName(&pExpr->pRight->colInfo, pQueryInfo, &index) != TSDB_CODE_SUCCESS) { - return TSDB_CODE_TSC_INVALID_SQL; + return invalidSqlErrMsg(pQueryInfo->msg, msg4); } pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); - tagColIndex = index.columnIndex - tscGetNumOfColumns(pTableMetaInfo->pTableMeta); + SSchema* pTagSchema2 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index.columnIndex); pRight->uid = pTableMetaInfo->pTableMeta->uid; - pRight->tagCol = tagColIndex; + pRight->tagColId = pTagSchema2->colId; strcpy(pRight->tableId, pTableMetaInfo->name); + if (pTagSchema1->type != pTagSchema2->type) { + return invalidSqlErrMsg(pQueryInfo->msg, msg3); + } + + if (pTagSchema1->type == TSDB_DATA_TYPE_BINARY || pTagSchema1->type == TSDB_DATA_TYPE_NCHAR) { + return invalidSqlErrMsg(pQueryInfo->msg, msg2); + } + pTagCond->joinInfo.hasJoin = true; return TSDB_CODE_SUCCESS; } @@ -3815,6 +3764,10 @@ static int32_t getTagQueryCondExpr(SQueryInfo* pQueryInfo, SCondExpr* pCondExpr, for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) { tSQLExpr* p1 = extractExprForSTable(pExpr, pQueryInfo, i); + if (p1 == NULL) { // no query condition on this table + continue; + } + tExprNode* p = NULL; SArray* colList = taosArrayInit(10, sizeof(SColIndex)); @@ -4948,25 +4901,25 @@ int32_t parseCreateDBOptions(SSqlCmd* pCmd, SCreateDBInfo* pCreateDbSql) { return TSDB_CODE_SUCCESS; } -void tscAddTimestampColumn(SQueryInfo* pQueryInfo, int16_t functionId, int16_t tableIndex) { - // the first column not timestamp column, add it - SSqlExpr* pExpr = NULL; - if (tscSqlExprNumOfExprs(pQueryInfo) > 0) { - pExpr = tscSqlExprGet(pQueryInfo, 0); - } - - if (pExpr == NULL || pExpr->colInfo.colId != PRIMARYKEY_TIMESTAMP_COL_INDEX || pExpr->functionId != functionId) { - SColumnIndex index = {tableIndex, PRIMARYKEY_TIMESTAMP_COL_INDEX}; - - pExpr = tscSqlExprInsert(pQueryInfo, 0, functionId, &index, TSDB_DATA_TYPE_TIMESTAMP, TSDB_KEYSIZE, TSDB_KEYSIZE, false); - pExpr->colInfo.flag = TSDB_COL_NORMAL; - - // NOTE: tag column does not add to source column list - SColumnList ids = getColumnList(1, tableIndex, PRIMARYKEY_TIMESTAMP_COL_INDEX); - - insertResultField(pQueryInfo, 0, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, "ts", pExpr); - } -} +//void tscAddTimestampColumn(SQueryInfo* pQueryInfo, int16_t functionId, int16_t tableIndex) { +// // the first column not timestamp column, add it +// SSqlExpr* pExpr = NULL; +// if (tscSqlExprNumOfExprs(pQueryInfo) > 0) { +// pExpr = tscSqlExprGet(pQueryInfo, 0); +// } +// +// if (pExpr == NULL || pExpr->colInfo.colId != PRIMARYKEY_TIMESTAMP_COL_INDEX || pExpr->functionId != functionId) { +// SColumnIndex index = {tableIndex, PRIMARYKEY_TIMESTAMP_COL_INDEX}; +// +// pExpr = tscSqlExprInsert(pQueryInfo, 0, functionId, &index, TSDB_DATA_TYPE_TIMESTAMP, TSDB_KEYSIZE, TSDB_KEYSIZE, false); +// pExpr->colInfo.flag = TSDB_COL_NORMAL; +// +// // NOTE: tag column does not add to source column list +// SColumnList ids = getColumnList(1, tableIndex, PRIMARYKEY_TIMESTAMP_COL_INDEX); +// +// insertResultField(pQueryInfo, 0, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, "ts", pExpr); +// } +//} void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClauseIndex, int32_t tableIndex) { SQueryInfo* pParentQueryInfo = tscGetQueryInfoDetail(&pParentObj->cmd, subClauseIndex); @@ -4979,7 +4932,7 @@ void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClau if (pExpr->functionId != TSDB_FUNC_TAG) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex); - int16_t columnInfo = tscGetJoinTagColIndexByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid); + int16_t columnInfo = tscGetJoinTagColIdByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid); SColumnIndex index = {.tableIndex = 0, .columnIndex = columnInfo}; SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta); @@ -5015,27 +4968,17 @@ static void doLimitOutputNormalColOfGroupby(SSqlExpr* pExpr) { void doAddGroupColumnForSubquery(SQueryInfo* pQueryInfo, int32_t tagIndex) { SColIndex* pColIndex = taosArrayGet(pQueryInfo->groupbyExpr.columnInfo, tagIndex); - int32_t index = pColIndex->colIndex; - + size_t size = tscSqlExprNumOfExprs(pQueryInfo); + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); - SSchema* pSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index); - SColumnIndex colIndex = {.tableIndex = 0, .columnIndex = index}; - - size_t size = tscSqlExprNumOfExprs(pQueryInfo); - SSqlExpr* pExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_PRJ, &colIndex, pSchema->type, pSchema->bytes, - pSchema->bytes, false); + SSchema* pSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pColIndex->colIndex); + SColumnIndex colIndex = {.tableIndex = 0, .columnIndex = pColIndex->colIndex}; - pExpr->colInfo.flag = TSDB_COL_NORMAL; - doLimitOutputNormalColOfGroupby(pExpr); + tscAddSpecialColumnForSelect(pQueryInfo, size, TSDB_FUNC_PRJ, &colIndex, pSchema, TSDB_COL_NORMAL); - // NOTE: tag column does not add to source column list - SColumnList list = {0}; - list.num = 1; - list.ids[0] = colIndex; - - insertResultField(pQueryInfo, size, &list, pSchema->bytes, pSchema->type, pSchema->name, pExpr); - SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pQueryInfo->fieldsInfo, size - 1); + SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pQueryInfo->fieldsInfo, size); + doLimitOutputNormalColOfGroupby(pInfo->pSqlExpr); pInfo->visible = false; } @@ -5247,6 +5190,7 @@ static int32_t doAddGroupbyColumnsOnDemand(SQueryInfo* pQueryInfo) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + SSchema s = tGetTableNameColumnSchema(); SSchema* pSchema = tscGetTableSchema(pTableMetaInfo->pTableMeta); int16_t bytes = 0; int16_t type = 0; @@ -5254,10 +5198,8 @@ static int32_t doAddGroupbyColumnsOnDemand(SQueryInfo* pQueryInfo) { for (int32_t i = 0; i < pQueryInfo->groupbyExpr.numOfGroupCols; ++i) { SColIndex* pColIndex = taosArrayGet(pQueryInfo->groupbyExpr.columnInfo, i); - SSchema s; int16_t colIndex = pColIndex->colIndex; if (colIndex == TSDB_TBNAME_COLUMN_INDEX) { - s = tGetTableNameColumnSchema(); type = s.type; bytes = s.bytes; name = s.name; @@ -5954,10 +5896,6 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) { setColumnOffsetValueInResultset(pQueryInfo); - for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) { - updateTagColumnIndex(pQueryInfo, i); - } - /* * fill options are set at the end position, when all columns are set properly * the columns may be increased due to group by operation diff --git a/src/client/src/tscSchemaUtil.c b/src/client/src/tscSchemaUtil.c index 439aa7c1de..0dfbf8c487 100644 --- a/src/client/src/tscSchemaUtil.c +++ b/src/client/src/tscSchemaUtil.c @@ -64,14 +64,6 @@ 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; } @@ -119,11 +111,24 @@ bool isValidSchema(struct SSchema* pSchema, int32_t numOfCols) { return (rowLen <= TSDB_MAX_BYTES_PER_ROW); } -SSchema* tscGetTableColumnSchema(const STableMeta* pTableMeta, int32_t startCol) { +SSchema* tscGetTableColumnSchema(const STableMeta* pTableMeta, int32_t colIndex) { assert(pTableMeta != NULL); SSchema* pSchema = (SSchema*) pTableMeta->schema; - return &pSchema[startCol]; + return &pSchema[colIndex]; +} + +// TODO for large number of columns, employ the binary search method +SSchema* tscGetTableColumnSchemaById(STableMeta* pTableMeta, int16_t colId) { + STableComInfo tinfo = tscGetTableInfo(pTableMeta); + + for(int32_t i = 0; i < tinfo.numOfColumns + tinfo.numOfTags; ++i) { + if (pTableMeta->schema[i].colId == colId) { + return &pTableMeta->schema[i]; + } + } + + return NULL; } struct SSchema tscGetTbnameColumnSchema() { diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index c4785b377c..5a2054bbcd 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -14,20 +14,18 @@ */ #include "os.h" +#include "qsqltype.h" #include "tcache.h" #include "trpc.h" +#include "tscLocalMerge.h" +#include "tscLog.h" #include "tscProfile.h" -#include "tscSecondaryMerge.h" -#include "tscSubquery.h" #include "tscUtil.h" #include "tschemautil.h" #include "tsclient.h" -#include "tsocket.h" #include "ttime.h" #include "ttimer.h" #include "tutil.h" -#include "tscLog.h" -#include "qsqltype.h" #define TSC_MGMT_VNODE 999 @@ -644,7 +642,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { pQueryMsg->numOfGroupCols = htons(pQueryInfo->groupbyExpr.numOfGroupCols); pQueryMsg->numOfTags = htonl(numOfTags); pQueryMsg->tagNameRelType = htons(pQueryInfo->tagCond.relType); - pQueryMsg->queryType = htons(pQueryInfo->type); + pQueryMsg->queryType = htonl(pQueryInfo->type); size_t numOfOutput = tscSqlExprNumOfExprs(pQueryInfo); pQueryMsg->numOfOutput = htons(numOfOutput); @@ -723,6 +721,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { pMsg += sizeof(SSqlFuncMsg); for (int32_t j = 0; j < pExpr->numOfParams; ++j) { + // todo add log pSqlFuncExpr->arg[j].argType = htons((uint16_t)pExpr->param[j].nType); pSqlFuncExpr->arg[j].argBytes = htons(pExpr->param[j].nLen); @@ -800,6 +799,27 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { } } + // serialize tag column query condition + if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0) { + STagCond* pTagCond = &pQueryInfo->tagCond; + + SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->uid); + if (pCond != NULL && pCond->cond != NULL) { + pQueryMsg->tagCondLen = htons(pCond->len); + memcpy(pMsg, pCond->cond, pCond->len); + + pMsg += pCond->len; + } + } + + if (pQueryInfo->tagCond.tbnameCond.cond == NULL) { + *pMsg = 0; + pMsg++; + } else { + strcpy(pMsg, pQueryInfo->tagCond.tbnameCond.cond); + pMsg += strlen(pQueryInfo->tagCond.tbnameCond.cond) + 1; + } + // compressed ts block pQueryMsg->tsOffset = htonl(pMsg - pStart); int32_t tsLen = 0; @@ -824,27 +844,6 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { pQueryMsg->tsOrder = htonl(pQueryInfo->tsBuf->tsOrder); } - // serialize tag column query condition - if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0) { - STagCond* pTagCond = &pQueryInfo->tagCond; - - SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->uid); - if (pCond != NULL && pCond->cond != NULL) { - pQueryMsg->tagCondLen = htons(pCond->len); - memcpy(pMsg, pCond->cond, pCond->len); - - pMsg += pCond->len; - } - } - - if (pQueryInfo->tagCond.tbnameCond.cond == NULL) { - *pMsg = 0; - pMsg++; - } else { - strcpy(pMsg, pQueryInfo->tagCond.tbnameCond.cond); - pMsg += strlen(pQueryInfo->tagCond.tbnameCond.cond) + 1; - } - int32_t msgLen = pMsg - pStart; tscTrace("%p msg built success,len:%d bytes", pSql, msgLen); @@ -1175,7 +1174,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) { strcpy(pCreateTableMsg->tableId, pTableMetaInfo->name); // use dbinfo from table id without modifying current db info - tscGetDBInfoFromMeterId(pTableMetaInfo->name, pCreateTableMsg->db); + tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pCreateTableMsg->db); SCreateTableSQL *pCreateTable = pInfo->pCreateTableInfo; @@ -1252,7 +1251,7 @@ int tscBuildAlterTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) { } SCMAlterTableMsg *pAlterTableMsg = (SCMAlterTableMsg *)pCmd->payload; - tscGetDBInfoFromMeterId(pTableMetaInfo->name, pAlterTableMsg->db); + tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pAlterTableMsg->db); strcpy(pAlterTableMsg->tableId, pTableMetaInfo->name); pAlterTableMsg->type = htons(pAlterInfo->type); @@ -1577,7 +1576,7 @@ int tscBuildSTableVgroupMsg(SSqlObj *pSql, SSqlInfo *pInfo) { pMsg = pStart; SMgmtHead *pMgmt = (SMgmtHead *)pMsg; - tscGetDBInfoFromMeterId(pTableMetaInfo->name, pMgmt->db); + tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pMgmt->db); pMsg += sizeof(SMgmtHead); @@ -1932,113 +1931,6 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) { } int tscProcessSTableVgroupRsp(SSqlObj *pSql) { -#if 0 - void ** metricMetaList = NULL; - int32_t * sizes = NULL; - - int32_t num = htons(*(int16_t *)rsp); - rsp += sizeof(int16_t); - - metricMetaList = calloc(1, POINTER_BYTES * num); - sizes = calloc(1, sizeof(int32_t) * num); - - // return with error code - if (metricMetaList == NULL || sizes == NULL) { - tfree(metricMetaList); - tfree(sizes); - pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY; - - return pSql->res.code; - } - - for (int32_t k = 0; k < num; ++k) { - pMeta = (SSuperTableMeta *)rsp; - - size_t size = (size_t)pSql->res.rspLen - 1; - rsp = rsp + sizeof(SSuperTableMeta); - - pMeta->numOfTables = htonl(pMeta->numOfTables); - pMeta->numOfVnodes = htonl(pMeta->numOfVnodes); - pMeta->tagLen = htons(pMeta->tagLen); - - size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfTables * sizeof(STableIdInfo *); - - char *pBuf = calloc(1, size); - if (pBuf == NULL) { - pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY; - goto _error_clean; - } - - SSuperTableMeta *pNewMetricMeta = (SSuperTableMeta *)pBuf; - metricMetaList[k] = pNewMetricMeta; - - pNewMetricMeta->numOfTables = pMeta->numOfTables; - pNewMetricMeta->numOfVnodes = pMeta->numOfVnodes; - pNewMetricMeta->tagLen = pMeta->tagLen; - - pBuf = pBuf + sizeof(SSuperTableMeta) + pNewMetricMeta->numOfVnodes * sizeof(SVnodeSidList *); - - for (int32_t i = 0; i < pMeta->numOfVnodes; ++i) { - SVnodeSidList *pSidLists = (SVnodeSidList *)rsp; - memcpy(pBuf, pSidLists, sizeof(SVnodeSidList)); - - pNewMetricMeta->list[i] = pBuf - (char *)pNewMetricMeta; // offset value - SVnodeSidList *pLists = (SVnodeSidList *)pBuf; - - tscTrace("%p metricmeta:vid:%d,numOfTables:%d", pSql, i, pLists->numOfSids); - - pBuf += sizeof(SVnodeSidList) + sizeof(STableIdInfo *) * pSidLists->numOfSids; - rsp += sizeof(SVnodeSidList); - - size_t elemSize = sizeof(STableIdInfo) + pNewMetricMeta->tagLen; - for (int32_t j = 0; j < pSidLists->numOfSids; ++j) { - pLists->pSidExtInfoList[j] = pBuf - (char *)pLists; - memcpy(pBuf, rsp, elemSize); - - ((STableIdInfo *)pBuf)->uid = htobe64(((STableIdInfo *)pBuf)->uid); - ((STableIdInfo *)pBuf)->sid = htonl(((STableIdInfo *)pBuf)->sid); - - rsp += elemSize; - pBuf += elemSize; - } - } - - sizes[k] = pBuf - (char *)pNewMetricMeta; - } - - SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0); - for (int32_t i = 0; i < num; ++i) { - char name[TSDB_MAX_TAGS_LEN + 1] = {0}; - - STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, i); - tscGetMetricMetaCacheKey(pQueryInfo, name, pTableMetaInfo->pTableMeta->uid); - -#ifdef _DEBUG_VIEW - printf("generate the metric key:%s, index:%d\n", name, i); -#endif - - // release the used metricmeta - taosCacheRelease(tscCacheHandle, (void **)&(pTableMetaInfo->pMetricMeta), false); - pTableMetaInfo->pMetricMeta = (SSuperTableMeta *)taosCachePut(tscCacheHandle, name, (char *)metricMetaList[i], - sizes[i], tsMetricMetaKeepTimer); - tfree(metricMetaList[i]); - - // failed to put into cache - if (pTableMetaInfo->pMetricMeta == NULL) { - pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY; - goto _error_clean; - } - } - -_error_clean: - // free allocated resource - for (int32_t i = 0; i < num; ++i) { - tfree(metricMetaList[i]); - } - - free(sizes); - free(metricMetaList); -#endif SSqlRes* pRes = &pSql->res; // NOTE: the order of several table must be preserved. diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 6f043f186a..126e34704c 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -20,14 +20,9 @@ #include "tnote.h" #include "trpc.h" #include "tscLog.h" -#include "tscProfile.h" -#include "tscSecondaryMerge.h" #include "tscSubquery.h" #include "tscUtil.h" #include "tsclient.h" -#include "tscompression.h" -#include "tsocket.h" -#include "ttimer.h" #include "ttokendef.h" #include "tutil.h" diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index 1a1305825b..7e3aaf7fc5 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -238,7 +238,7 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) { if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { taosArraySort( tables, tscCompareTidTags ); - tscBuildVgroupTableInfo( pTableMetaInfo, tables ); + tscBuildVgroupTableInfo(pSql, pTableMetaInfo, tables); } taosArrayDestroy(tables); diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index 3bc931a855..4e02325be6 100644 --- a/src/client/src/tscSubquery.c +++ b/src/client/src/tscSubquery.c @@ -28,7 +28,7 @@ typedef struct SInsertSupporter { } SInsertSupporter; static void freeJoinSubqueryObj(SSqlObj* pSql); -static bool tscHashRemainDataInSubqueryResultSet(SSqlObj *pSql); +static bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql); static bool tsCompare(int32_t order, int64_t left, int64_t right) { if (order == TSDB_ORDER_ASC) { @@ -38,16 +38,15 @@ static bool tsCompare(int32_t order, int64_t left, int64_t right) { } } -static int64_t doTSBlockIntersect(SSqlObj* pSql, SJoinSupporter* pSupporter1, - SJoinSupporter* pSupporter2, TSKEY* st, TSKEY* et) { - STSBuf* output1 = tsBufCreate(true); - STSBuf* output2 = tsBufCreate(true); - - *st = INT64_MAX; - *et = INT64_MIN; - +static int64_t doTSBlockIntersect(SSqlObj* pSql, SJoinSupporter* pSupporter1, SJoinSupporter* pSupporter2, STimeWindow * win) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex); - + + STSBuf* output1 = tsBufCreate(true, pQueryInfo->order.order); + STSBuf* output2 = tsBufCreate(true, pQueryInfo->order.order); + + win->skey = INT64_MAX; + win->ekey = INT64_MIN; + SLimitVal* pLimit = &pQueryInfo->limit; int32_t order = pQueryInfo->order.order; @@ -106,12 +105,12 @@ static int64_t doTSBlockIntersect(SSqlObj* pSql, SJoinSupporter* pSupporter1, * final results which is acquired after the secondry merge of in the client. */ if (pLimit->offset == 0 || pQueryInfo->intervalTime > 0 || QUERY_IS_STABLE_QUERY(pQueryInfo->type)) { - if (*st > elem1.ts) { - *st = elem1.ts; + if (win->skey > elem1.ts) { + win->skey = elem1.ts; } - if (*et < elem1.ts) { - *et = elem1.ts; + if (win->ekey < elem1.ts) { + win->ekey = elem1.ts; } tsBufAppend(output1, elem1.vnode, elem1.tag, (const char*)&elem1.ts, sizeof(elem1.ts)); @@ -151,8 +150,8 @@ static int64_t doTSBlockIntersect(SSqlObj* pSql, SJoinSupporter* pSupporter1, tsBufDestory(pSupporter2->pTSBuf); tscTrace("%p input1:%" PRId64 ", input2:%" PRId64 ", final:%" PRId64 " for secondary query after ts blocks " - "intersecting, skey:%" PRId64 ", ekey:%" PRId64, pSql, - numOfInput1, numOfInput2, output1->numOfTotal, *st, *et); + "intersecting, skey:%" PRId64 ", ekey:%" PRId64, pSql, numOfInput1, numOfInput2, output1->numOfTotal, + win->skey, win->ekey); return output1->numOfTotal; } @@ -252,7 +251,7 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) { assert(numOfSub > 0); // scan all subquery, if one sub query has only ts, ignore it - tscTrace("%p start to launch secondary subqueries, total:%d, only:%d needs to query, others are not retrieve in " + tscTrace("%p start to launch secondary subquery, total:%d, only:%d needs to query, others are not retrieve in " "select clause", pSql, pSql->numOfSubs, numOfSub); //the subqueries that do not actually launch the secondary query to virtual node is set as completed. @@ -301,7 +300,7 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) { // set the second stage sub query for join process TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE); - + pQueryInfo->intervalTime = pSupporter->interval; pQueryInfo->groupbyExpr = pSupporter->groupbyExpr; @@ -328,22 +327,38 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) { */ pSupporter->limit = pQueryInfo->limit; pNewQueryInfo->limit = pSupporter->limit; - - // fetch the join tag column + + SColumnIndex index = {.tableIndex = 0, .columnIndex = PRIMARYKEY_TIMESTAMP_COL_INDEX}; + SSchema* s = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, 0); + + SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0); + int16_t funcId = pExpr->functionId; + + if ((pExpr->colInfo.colId != PRIMARYKEY_TIMESTAMP_COL_INDEX) || + (funcId != TSDB_FUNC_TS && funcId != TSDB_FUNC_TS_DUMMY && funcId != TSDB_FUNC_PRJ)) { + + int16_t functionId = tscIsProjectionQuery(pQueryInfo)? TSDB_FUNC_PRJ : TSDB_FUNC_TS; + + tscAddSpecialColumnForSelect(pQueryInfo, 0, functionId, &index, s, TSDB_COL_NORMAL); + tscPrintSelectClause(pNew, 0); + tscFieldInfoUpdateOffset(pNewQueryInfo); + + pExpr = tscSqlExprGet(pQueryInfo, 0); + } + + // set the join condition tag column info, to do extract method if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { - SSqlExpr* pExpr = tscSqlExprGet(pNewQueryInfo, 0); assert(pQueryInfo->tagCond.joinInfo.hasJoin); - - int16_t tagColIndex = tscGetJoinTagColIndexByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid); - pExpr->param[0].i64Key = tagColIndex; + int16_t colId = tscGetJoinTagColIdByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid); + + pExpr->param[0].i64Key = colId; pExpr->numOfParams = 1; } - + size_t numOfCols = taosArrayGetSize(pNewQueryInfo->colList); tscTrace("%p subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s", - pSql, pNew, 0, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type, - taosArrayGetSize(pNewQueryInfo->exprList), numOfCols, - pNewQueryInfo->fieldsInfo.numOfOutput, pNewQueryInfo->pTableMetaInfo[0]->name); + pSql, pNew, 0, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type, taosArrayGetSize(pNewQueryInfo->exprList), + numOfCols, pNewQueryInfo->fieldsInfo.numOfOutput, pTableMetaInfo->name); } //prepare the subqueries object failed, abort @@ -357,12 +372,10 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) { } for(int32_t i = 0; i < pSql->numOfSubs; ++i) { - SSqlObj* pSub = pSql->pSubs[i]; - if (pSub == NULL) { + if (pSql->pSubs[i] == NULL) { continue; } - - tscProcessSql(pSub); + tscDoQuery(pSql->pSubs[i]); } return TSDB_CODE_SUCCESS; @@ -403,11 +416,9 @@ static void quitAllSubquery(SSqlObj* pSqlObj, SJoinSupporter* pSupporter) { } // update the query time range according to the join results on timestamp -static void updateQueryTimeRange(SQueryInfo* pQueryInfo, int64_t st, int64_t et) { - assert(pQueryInfo->window.skey <= st && pQueryInfo->window.ekey >= et); - - pQueryInfo->window.skey = st; - pQueryInfo->window.ekey = et; +static void updateQueryTimeRange(SQueryInfo* pQueryInfo, STimeWindow* win) { + assert(pQueryInfo->window.skey <= win->skey && pQueryInfo->window.ekey >= win->ekey); + pQueryInfo->window = *win; } static void tSIntersectionAndLaunchSecQuery(SJoinSupporter* pSupporter, SSqlObj* pSql) { @@ -451,21 +462,21 @@ static void tSIntersectionAndLaunchSecQuery(SJoinSupporter* pSupporter, SSqlObj* SJoinSupporter* p1 = pParentSql->pSubs[0]->param; SJoinSupporter* p2 = pParentSql->pSubs[1]->param; - TSKEY st, et; - int64_t num = doTSBlockIntersect(pParentSql, p1, p2, &st, &et); + STimeWindow win = TSWINDOW_INITIALIZER; + int64_t num = doTSBlockIntersect(pParentSql, p1, p2, &win); if (num <= 0) { // no result during ts intersect tscTrace("%p free all sub SqlObj and quit", pParentSql); freeJoinSubqueryObj(pParentSql); } else { - updateQueryTimeRange(pParentQueryInfo, st, et); + updateQueryTimeRange(pParentQueryInfo, &win); tscLaunchSecondPhaseSubqueries(pParentSql); } } } int32_t tscCompareTidTags(const void* p1, const void* p2) { - const STidTags* t1 = (const STidTags*) p1; - const STidTags* t2 = (const STidTags*) p2; + const STidTags* t1 = (const STidTags*) varDataVal(p1); + const STidTags* t2 = (const STidTags*) varDataVal(p2); if (t1->vgId != t2->vgId) { return (t1->vgId > t2->vgId) ? 1 : -1; @@ -476,38 +487,40 @@ int32_t tscCompareTidTags(const void* p1, const void* p2) { return 0; } -void tscBuildVgroupTableInfo(STableMetaInfo* pTableMetaInfo, SArray* tables) { - SArray* result = taosArrayInit( 4, sizeof(SVgroupTableInfo) ); - SArray* vgTables = NULL; +void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables) { + SArray* result = taosArrayInit(4, sizeof(SVgroupTableInfo)); + SArray* vgTables = NULL; STidTags* prev = NULL; - size_t numOfTables = taosArrayGetSize( tables ); - for( size_t i = 0; i < numOfTables; i++ ) { - STidTags* tt = taosArrayGet( tables, i ); + size_t numOfTables = taosArrayGetSize(tables); + for (size_t i = 0; i < numOfTables; i++) { + STidTags* tt = taosArrayGet(tables, i); - if( prev == NULL || tt->vgId != prev->vgId ) { + if (prev == NULL || tt->vgId != prev->vgId) { SVgroupsInfo* pvg = pTableMetaInfo->vgroupList; - SVgroupTableInfo info = {{ 0 }}; - for( int32_t m = 0; m < pvg->numOfVgroups; ++m ) { - if( tt->vgId == pvg->vgroups[m].vgId ) { + SVgroupTableInfo info = {{0}}; + for (int32_t m = 0; m < pvg->numOfVgroups; ++m) { + if (tt->vgId == pvg->vgroups[m].vgId) { info.vgInfo = pvg->vgroups[m]; break; } } - assert( info.vgInfo.numOfIps != 0 ); + assert(info.vgInfo.numOfIps != 0); - vgTables = taosArrayInit( 4, sizeof(STableIdInfo) ); + vgTables = taosArrayInit(4, sizeof(STableIdInfo)); info.itemList = vgTables; - taosArrayPush( result, &info ); + taosArrayPush(result, &info); } - STableIdInfo item = { .uid = tt->uid, .tid = tt->tid, .key = INT64_MIN }; - taosArrayPush( vgTables, &item ); + tscTrace("%p tid:%d, uid:%"PRIu64",vgId:%d added for vnode query", pSql, tt->tid, tt->uid, tt->vgId) + STableIdInfo item = {.uid = tt->uid, .tid = tt->tid, .key = INT64_MIN}; + taosArrayPush(vgTables, &item); prev = tt; } pTableMetaInfo->pVgroupTables = result; + pTableMetaInfo->vgroupIndex = 0; } static void issueTSCompQuery(SSqlObj* pSql, SJoinSupporter* pSupporter, SSqlObj* pParent) { @@ -533,9 +546,8 @@ static void issueTSCompQuery(SSqlObj* pSql, SJoinSupporter* pSupporter, SSqlObj* // set the tags value for ts_comp function SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, 0); - int16_t tagColIndex = tscGetJoinTagColIndexByUid(&pSupporter->tagCond, pTableMetaInfo->pTableMeta->uid); - - pExpr->param->i64Key = tagColIndex; + int16_t tagColId = tscGetJoinTagColIdByUid(&pSupporter->tagCond, pTableMetaInfo->pTableMeta->uid); + pExpr->param->i64Key = tagColId; pExpr->numOfParams = 1; // add the filter tag column @@ -555,127 +567,198 @@ static void issueTSCompQuery(SSqlObj* pSql, SJoinSupporter* pSupporter, SSqlObj* size_t numOfCols = taosArrayGetSize(pQueryInfo->colList); tscTrace( - "%p subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, transfer to ts_comp query to retrieve timestamps, " - "exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s", - pParent, pSql, 0, pTableMetaInfo->vgroupIndex, pQueryInfo->type, tscSqlExprNumOfExprs(pQueryInfo), - numOfCols, pQueryInfo->fieldsInfo.numOfOutput, pTableMetaInfo->name); + "%p subquery:%p tableIndex:%d, vgroupIndex:%d, numOfVgroups:%d, type:%d, ts_comp query to retrieve timestamps, " + "numOfExpr:%d, colList:%d, numOfOutputFields:%d, name:%s", + pParent, pSql, 0, pTableMetaInfo->vgroupIndex, pTableMetaInfo->vgroupList->numOfVgroups, pQueryInfo->type, + tscSqlExprNumOfExprs(pQueryInfo), numOfCols, pQueryInfo->fieldsInfo.numOfOutput, pTableMetaInfo->name); tscProcessSql(pSql); } +static bool checkForIdenticalTagVal(SQueryInfo* pQueryInfo, SJoinSupporter* p1, void* pSql) { + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + + SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta);// todo: tags mismatch, tags not completed + SColumn *pCol = taosArrayGetP(pTableMetaInfo->tagColList, 0); + SSchema *pColSchema = &pSchema[pCol->colIndex.columnIndex]; + + for(int32_t i = 1; i < p1->num; ++i) { + STidTags* prev = (STidTags*) varDataVal(p1->pIdTagList + (i - 1) * p1->tagSize); + STidTags* p = (STidTags*) varDataVal(p1->pIdTagList + i * p1->tagSize); + + if (doCompare(prev->tag, p->tag, pColSchema->type, pColSchema->bytes) == 0) { + tscError("%p join tags have same value for different table, free all sub SqlObj and quit", pSql); + p1->pState->code = TSDB_CODE_QRY_DUP_JOIN_KEY; + return false; + } + } + + return true; +} + +static void getIntersectionOfTagVal(SQueryInfo* pQueryInfo, SSqlObj* pParentSql, SArray** s1, SArray** s2) { + tscTrace("%p all subqueries retrieve tags complete, do tags match", pParentSql); + + SJoinSupporter* p1 = pParentSql->pSubs[0]->param; + SJoinSupporter* p2 = pParentSql->pSubs[1]->param; + + qsort(p1->pIdTagList, p1->num, p1->tagSize, tscCompareTidTags); + qsort(p2->pIdTagList, p2->num, p2->tagSize, tscCompareTidTags); + + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + int16_t tagColId = tscGetJoinTagColIdByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->uid); + + SSchema* pColSchema = tscGetTableColumnSchemaById(pTableMetaInfo->pTableMeta, tagColId); + + *s1 = taosArrayInit(p1->num, p1->tagSize); + *s2 = taosArrayInit(p2->num, p2->tagSize); + + if (!(checkForIdenticalTagVal(pQueryInfo, p1, pParentSql) && checkForIdenticalTagVal(pQueryInfo, p2, pParentSql))) { + freeJoinSubqueryObj(pParentSql); + pParentSql->res.code = TSDB_CODE_QRY_DUP_JOIN_KEY; + tscQueueAsyncRes(pParentSql); + return; + } + + int32_t i = 0, j = 0; + while(i < p1->num && j < p2->num) { + STidTags* pp1 = (STidTags*) varDataVal(p1->pIdTagList + i * p1->tagSize); + STidTags* pp2 = (STidTags*) varDataVal(p2->pIdTagList + j * p2->tagSize); + + int32_t ret = doCompare(pp1->tag, pp2->tag, pColSchema->type, pColSchema->bytes); + if (ret == 0) { + tscTrace("%p tag matched, vgId:%d, val:%d, tid:%d, uid:%"PRIu64", tid:%d, uid:%"PRIu64, pParentSql, pp1->vgId, + *(int*) pp1->tag, pp1->tid, pp1->uid, pp2->tid, pp2->uid); + + taosArrayPush(*s1, pp1); + taosArrayPush(*s2, pp2); + j++; + i++; + } else if (ret > 0) { + j++; + } else { + i++; + } + } +} + static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) { SJoinSupporter* pSupporter = (SJoinSupporter*)param; - + SSqlObj* pParentSql = pSupporter->pObj; + SSqlObj* pSql = (SSqlObj*)tres; SSqlCmd* pCmd = &pSql->cmd; - - SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); - + SSqlRes* pRes = &pSql->res; + + SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); + // response of tag retrieve if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_TAG_FILTER_QUERY)) { - if (numOfRows == 0 || pSql->res.completed) { - + // todo handle error + + if (numOfRows == 0 || pRes->completed) { if (numOfRows > 0) { - size_t length = pSupporter->totalLen + pSql->res.rspLen; - char* tmp = realloc(pSupporter->pIdTagList, length); + size_t validLen = pSupporter->tagSize * pRes->numOfRows; + + size_t length = pSupporter->totalLen + validLen; + char* tmp = realloc(pSupporter->pIdTagList, length); assert(tmp != NULL); pSupporter->pIdTagList = tmp; - - memcpy(pSupporter->pIdTagList, pSql->res.data, pSql->res.rspLen); - pSupporter->totalLen += pSql->res.rspLen; - pSupporter->num += pSql->res.numOfRows; + + memcpy(pSupporter->pIdTagList + pSupporter->totalLen,pRes->data, validLen); + pSupporter->totalLen += validLen; + pSupporter->num += pRes->numOfRows; } - + + // tuples have been retrieved to client, try tuples from the next vnode + if (hasMoreVnodesToTry(pSql)) { + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + + int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; + pTableMetaInfo->vgroupIndex += 1; + assert(pTableMetaInfo->vgroupIndex < totalVgroups); + + tscTrace("%p tid_tag from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%d", + pSql, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, + pSupporter->num); + + pCmd->command = TSDB_SQL_SELECT; + tscResetForNextRetrieve(&pSql->res); + + // set the callback function + pSql->fp = tscJoinQueryCallback; + tscProcessSql(pSql); + return; + } + int32_t numOfTotal = pSupporter->pState->numOfTotal; int32_t finished = atomic_add_fetch_32(&pSupporter->pState->numOfCompleted, 1); - if (finished < numOfTotal) { return; } - - // all subqueries are returned, start to compare the tags + + // all subquery are returned, start to compare the tags assert(finished == numOfTotal); - tscTrace("%p all subqueries retrieve tags complete, do tags match", pParentSql); - - SJoinSupporter* p1 = pParentSql->pSubs[0]->param; - SJoinSupporter* p2 = pParentSql->pSubs[1]->param; - - qsort(p1->pIdTagList, p1->num, p1->tagSize, tscCompareTidTags); - qsort(p2->pIdTagList, p2->num, p2->tagSize, tscCompareTidTags); - - STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); - - SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta);// todo: tags mismatch, tags not completed - - SColumn *pCol = taosArrayGetP(pTableMetaInfo->tagColList, 0); - SSchema *pColSchema = &pSchema[pCol->colIndex.columnIndex]; - - SArray* s1 = taosArrayInit(p1->num, p1->tagSize); - SArray* s2 = taosArrayInit(p2->num, p2->tagSize); - - int32_t i = 0, j = 0; - while(i < p1->num && j < p2->num) { - STidTags* pp1 = (STidTags*) p1->pIdTagList + i * p1->tagSize; - STidTags* pp2 = (STidTags*) p2->pIdTagList + j * p2->tagSize; - - int32_t ret = doCompare(pp1->tag, pp2->tag, pColSchema->type, pColSchema->bytes); - if (ret == 0) { - taosArrayPush(s1, pp1); - taosArrayPush(s2, pp2); - j++; - i++; - } else if (ret > 0) { - j++; - } else { - i++; - } - } - - if (taosArrayGetSize(s1) == 0 || taosArrayGetSize(s2) == 0) {// no results,return. + + SArray *s1 = NULL, *s2 = NULL; + getIntersectionOfTagVal(pQueryInfo, pParentSql, &s1, &s2); + + if (taosArrayGetSize(s1) == 0 || taosArrayGetSize(s2) == 0) { // no results,return. tscTrace("%p free all sub SqlObj and quit", pParentSql); freeJoinSubqueryObj(pParentSql); return; } else { SSqlCmd* pSubCmd1 = &pParentSql->pSubs[0]->cmd; SSqlCmd* pSubCmd2 = &pParentSql->pSubs[1]->cmd; - - SQueryInfo* pQueryInfo1 = tscGetQueryInfoDetail(pSubCmd1, 0); + + SQueryInfo* pQueryInfo1 = tscGetQueryInfoDetail(pSubCmd1, 0); STableMetaInfo* pTableMetaInfo1 = tscGetMetaInfo(pQueryInfo1, 0); - tscBuildVgroupTableInfo(pTableMetaInfo1, s1); - - SQueryInfo* pQueryInfo2 = tscGetQueryInfoDetail(pSubCmd2, 0); + tscBuildVgroupTableInfo(pParentSql, pTableMetaInfo1, s1); + + SQueryInfo* pQueryInfo2 = tscGetQueryInfoDetail(pSubCmd2, 0); STableMetaInfo* pTableMetaInfo2 = tscGetMetaInfo(pQueryInfo2, 0); - tscBuildVgroupTableInfo(pTableMetaInfo2, s2); - + tscBuildVgroupTableInfo(pParentSql, pTableMetaInfo2, s2); + pSupporter->pState->numOfCompleted = 0; pSupporter->pState->code = 0; pSupporter->pState->numOfTotal = 2; - - for(int32_t m = 0; m < pParentSql->numOfSubs; ++m) { + + for (int32_t m = 0; m < pParentSql->numOfSubs; ++m) { SSqlObj* psub = pParentSql->pSubs[m]; issueTSCompQuery(psub, psub->param, pParentSql); } } - + } else { - size_t length = pSupporter->totalLen + pSql->res.rspLen; + if (numOfRows < 0) { // error + pSupporter->pState->code = numOfRows; + quitAllSubquery(pParentSql, pSupporter); + + pParentSql->res.code = numOfRows; + tscQueueAsyncRes(pParentSql); + return; + } + + size_t length = pSupporter->totalLen + pRes->rspLen; + assert(length > 0); + char* tmp = realloc(pSupporter->pIdTagList, length); assert(tmp != NULL); - + pSupporter->pIdTagList = tmp; - - memcpy(pSupporter->pIdTagList, pSql->res.data, pSql->res.rspLen); - pSupporter->totalLen += pSql->res.rspLen; - pSupporter->num += pSql->res.numOfRows; - + + memcpy(pSupporter->pIdTagList, pRes->data, pRes->rspLen); + pSupporter->totalLen += pRes->rspLen; + pSupporter->num += pRes->numOfRows; + // continue retrieve data from vnode taos_fetch_rows_a(tres, joinRetrieveCallback, param); } - + return; } - + if (!TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { if (numOfRows < 0) { tscError("%p sub query failed, code:%s, index:%d", pSql, tstrerror(numOfRows), pSupporter->subqueryIndex); @@ -684,42 +767,64 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) { return; } - if (numOfRows == 0) { - tSIntersectionAndLaunchSecQuery(pSupporter, pSql); - return; + if (numOfRows > 0) { // write the compressed timestamp to disk file + fwrite(pRes->data, pRes->numOfRows, 1, pSupporter->f); + fclose(pSupporter->f); + pSupporter->f = NULL; + + STSBuf* pBuf = tsBufCreateFromFile(pSupporter->path, true); + if (pBuf == NULL) { // in error process, close the fd + tscError("%p invalid ts comp file from vnode, abort subquery, file size:%d", pSql, numOfRows); + + pSupporter->pState->code = TSDB_CODE_TSC_APP_ERROR; // todo set the informative code + quitAllSubquery(pParentSql, pSupporter); + return; + } + + if (pSupporter->pTSBuf == NULL) { + tscTrace("%p create tmp file for ts block:%s, size:%d bytes", pSql, pBuf->path, numOfRows); + pSupporter->pTSBuf = pBuf; + } else { + assert(pQueryInfo->numOfTables == 1); // for subquery, only one + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + + tsBufMerge(pSupporter->pTSBuf, pBuf, pTableMetaInfo->vgroupIndex); + tsBufDestory(pBuf); + } } - // write the compressed timestamp to disk file - fwrite(pSql->res.data, pSql->res.numOfRows, 1, pSupporter->f); - fclose(pSupporter->f); - pSupporter->f = NULL; - - STSBuf* pBuf = tsBufCreateFromFile(pSupporter->path, true); - if (pBuf == NULL) { - tscError("%p invalid ts comp file from vnode, abort subquery, file size:%d", pSql, numOfRows); + if (pRes->completed) { + if (hasMoreVnodesToTry(pSql)) { + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); - pSupporter->pState->code = TSDB_CODE_TSC_APP_ERROR; // todo set the informative code - quitAllSubquery(pParentSql, pSupporter); - return; - } + int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; + pTableMetaInfo->vgroupIndex += 1; + assert(pTableMetaInfo->vgroupIndex < totalVgroups); - if (pSupporter->pTSBuf == NULL) { - tscTrace("%p create tmp file for ts block:%s, size:%d bytes", pSql, pBuf->path, numOfRows); - pSupporter->pTSBuf = pBuf; - } else { - assert(pQueryInfo->numOfTables == 1); // for subquery, only one - STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + tscTrace("%p results from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%d", + pSql, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, + pRes->numOfClauseTotal); - tsBufMerge(pSupporter->pTSBuf, pBuf, pTableMetaInfo->vgroupIndex); - tsBufDestory(pBuf); - } - - if (pSql->res.completed) { - tSIntersectionAndLaunchSecQuery(pSupporter, pSql); - } else { // open a new file to save the incoming result + pCmd->command = TSDB_SQL_SELECT; + tscResetForNextRetrieve(&pSql->res); + + assert(pSupporter->f == NULL); + getTmpfilePath("ts-join", pSupporter->path); + pSupporter->f = fopen(pSupporter->path, "w"); + pRes->row = pRes->numOfRows; + + // set the callback function + pSql->fp = tscJoinQueryCallback; + tscProcessSql(pSql); + return; + } else { + tSIntersectionAndLaunchSecQuery(pSupporter, pSql); + } + + } else { // open a new file to save the incoming result getTmpfilePath("ts-join", pSupporter->path); pSupporter->f = fopen(pSupporter->path, "w"); - pSql->res.row = pSql->res.numOfRows; + pRes->row = pRes->numOfRows; taos_fetch_rows_a(tres, joinRetrieveCallback, param); } @@ -730,29 +835,29 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) { } if (numOfRows >= 0) { - pSql->res.numOfTotal += pSql->res.numOfRows; + pRes->numOfTotal += pRes->numOfRows; } - + if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && numOfRows == 0) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); assert(pQueryInfo->numOfTables == 1); - - // for projection query, need to try next vnode if current vnode is exhausted + + // for projection query, need to try next vnode if current vnode is exhausted if ((++pTableMetaInfo->vgroupIndex) < pTableMetaInfo->vgroupList->numOfVgroups) { 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; int32_t finished = atomic_add_fetch_32(&pSupporter->pState->numOfCompleted, 1); - + if (finished >= numOfTotal) { assert(finished == numOfTotal); tscTrace("%p all %d secondary subquery retrieves completed, global code:%d", tres, numOfTotal, @@ -763,17 +868,17 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) { freeJoinSubqueryObj(pParentSql); pParentSql->res.completed = true; } - + // update the records for each subquery in parent sql object. - for(int32_t i = 0; i < pParentSql->numOfSubs; ++i) { + for (int32_t i = 0; i < pParentSql->numOfSubs; ++i) { if (pParentSql->pSubs[i] == NULL) { continue; } - + SSqlRes* pRes1 = &pParentSql->pSubs[i]->res; pRes1->numOfClauseTotal += pRes1->numOfRows; } - + // data has retrieved to client, build the join results tscBuildResFromSubqueries(pParentSql); } else { @@ -833,7 +938,10 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) { } } } else { // has reach the limitation, no data anymore - hasData = false; + if (pRes->row >= pRes->numOfRows) { + hasData = false; + break; + } } } @@ -1055,10 +1163,8 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter // this data needs to be transfer to support struct memset(&pNewQueryInfo->fieldsInfo, 0, sizeof(SFieldInfo)); - - pSupporter->tagCond = pNewQueryInfo->tagCond; - memset(&pNewQueryInfo->tagCond, 0, sizeof(STagCond)); - + tscTagCondCopy(&pSupporter->tagCond, &pNewQueryInfo->tagCond);//pNewQueryInfo->tagCond; + pNew->cmd.numOfCols = 0; pNewQueryInfo->intervalTime = 0; memset(&pNewQueryInfo->limit, 0, sizeof(SLimitVal)); @@ -1071,39 +1177,34 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pNewQueryInfo, 0); if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { // return the tableId & tag - SSchema s = {0}; SColumnIndex index = {0}; - - size_t numOfTags = taosArrayGetSize(pTableMetaInfo->tagColList); - for(int32_t i = 0; i < numOfTags; ++i) { - SColumn* c = taosArrayGetP(pTableMetaInfo->tagColList, i); - index = (SColumnIndex) {.tableIndex = 0, .columnIndex = c->colIndex.columnIndex}; - - SSchema* pTagSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta); - s = pTagSchema[c->colIndex.columnIndex]; - - int16_t bytes = 0; - int16_t type = 0; - int32_t inter = 0; - - getResultDataInfo(s.type, s.bytes, TSDB_FUNC_TID_TAG, 0, &type, &bytes, &inter, 0, 0); - - s.type = type; - s.bytes = bytes; - pSupporter->tagSize = s.bytes; - } - + + STagCond* pTagCond = &pSupporter->tagCond; + assert(pTagCond->joinInfo.hasJoin); + + int32_t tagColId = tscGetJoinTagColIdByUid(pTagCond, pTableMetaInfo->pTableMeta->uid); + SSchema* s = tscGetTableColumnSchemaById(pTableMetaInfo->pTableMeta, tagColId); + + int16_t bytes = 0; + int16_t type = 0; + int32_t inter = 0; + + getResultDataInfo(s->type, s->bytes, TSDB_FUNC_TID_TAG, 0, &type, &bytes, &inter, 0, 0); + + SSchema s1 = {.colId = s->colId, .type = type, .bytes = bytes}; + pSupporter->tagSize = s1.bytes; + assert(isValidDataType(s1.type) && s1.bytes > 0); + // set get tags query type TSDB_QUERY_SET_TYPE(pNewQueryInfo->type, TSDB_QUERY_TYPE_TAG_FILTER_QUERY); - tscAddSpecialColumnForSelect(pNewQueryInfo, 0, TSDB_FUNC_TID_TAG, &index, &s, TSDB_COL_TAG); + tscAddSpecialColumnForSelect(pNewQueryInfo, 0, TSDB_FUNC_TID_TAG, &index, &s1, TSDB_COL_TAG); size_t numOfCols = taosArrayGetSize(pNewQueryInfo->colList); tscTrace( "%p subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, transfer to tid_tag query to retrieve (tableId, tags), " - "exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s", + "exprInfo:%d, colList:%d, fieldsInfo:%d, tagIndex:%d, name:%s", pSql, pNew, tableIndex, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type, tscSqlExprNumOfExprs(pNewQueryInfo), - numOfCols, pNewQueryInfo->fieldsInfo.numOfOutput, pNewQueryInfo->pTableMetaInfo[0]->name); - + numOfCols, pNewQueryInfo->fieldsInfo.numOfOutput, index.columnIndex, pNewQueryInfo->pTableMetaInfo[0]->name); } else { SSchema colSchema = {.type = TSDB_DATA_TYPE_BINARY, .bytes = 1}; SColumnIndex index = {0, PRIMARYKEY_TIMESTAMP_COL_INDEX}; @@ -1112,10 +1213,11 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter // set the tags value for ts_comp function SSqlExpr *pExpr = tscSqlExprGet(pNewQueryInfo, 0); - int16_t tagColIndex = tscGetJoinTagColIndexByUid(&pSupporter->tagCond, pTableMetaInfo->pTableMeta->uid); - - pExpr->param->i64Key = tagColIndex; - pExpr->numOfParams = 1; + if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { + int16_t tagColId = tscGetJoinTagColIdByUid(&pSupporter->tagCond, pTableMetaInfo->pTableMeta->uid); + pExpr->param->i64Key = tagColId; + pExpr->numOfParams = 1; + } // add the filter tag column if (pSupporter->colList != NULL) { @@ -1267,7 +1369,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) { pthread_mutexattr_t mutexattr; memset(&mutexattr, 0, sizeof(pthread_mutexattr_t)); - + pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_RECURSIVE_NP); pthread_mutex_init(&trs->queryMutex, &mutexattr); pthread_mutexattr_destroy(&mutexattr); @@ -1284,6 +1386,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) { if (pQueryInfo->tsBuf) { SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0); pNewQueryInfo->tsBuf = tsBufClone(pQueryInfo->tsBuf); + assert(pNewQueryInfo->tsBuf != NULL); } tscTrace("%p sub:%p create subquery success. orderOfSub:%d", pSql, pNew, trs->subqueryIndex); @@ -1437,7 +1540,7 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO // in case of second stage join subquery, invoke its callback function instead of regular QueueAsyncRes SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pPObj->cmd, 0); - if ((pQueryInfo->type & TSDB_QUERY_TYPE_JOIN_SEC_STAGE) == TSDB_QUERY_TYPE_JOIN_SEC_STAGE) { + if (!TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { (*pPObj->fp)(pPObj->param, pPObj, pPObj->res.code); } else { // regular super table query if (pPObj->res.code != TSDB_CODE_SUCCESS) { @@ -1458,7 +1561,7 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p // data in from current vnode is stored in cache and disk uint32_t numOfRowsFromSubquery = trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->num; - tscTrace("%p sub:%p all data retrieved from ip:%u,vgId:%d, numOfRows:%d, orderOfSub:%d", pPObj, pSql, + tscTrace("%p sub:%p all data retrieved from ip:%s, vgId:%d, numOfRows:%d, orderOfSub:%d", pPObj, pSql, pTableMetaInfo->vgroupList->vgroups[0].ipAddr[0].fqdn, pTableMetaInfo->vgroupList->vgroups[0].vgId, numOfRowsFromSubquery, idx); @@ -1692,7 +1795,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) { tscHandleSubqueryError(param, tres, pState->code); } else { // success, proceed to retrieve data from dnode - tscTrace("%p sub:%p query complete, ip:%u, vgId:%d, orderOfSub:%d, retrieve data", trsupport->pParentSqlObj, pSql, + tscTrace("%p sub:%p query complete, ip:%s, vgId:%d, orderOfSub:%d, retrieve data", trsupport->pParentSqlObj, pSql, pVgroup->ipAddr[0].fqdn, pVgroup->vgId, trsupport->subqueryIndex); if (pSql->res.qhandle == 0) { // qhandle is NULL, code is TSDB_CODE_SUCCESS means no results generated from this vnode @@ -1852,7 +1955,7 @@ void tscBuildResFromSubqueries(SSqlObj *pSql) { pRes->numOfClauseTotal++; break; } else { // continue retrieve data from vnode - if (!tscHashRemainDataInSubqueryResultSet(pSql)) { + if (!tscHasRemainDataInSubqueryResultSet(pSql)) { tscTrace("%p at least one subquery exhausted, free all other %d subqueries", pSql, pSql->numOfSubs - 1); SSubqueryState *pState = NULL; @@ -2002,7 +2105,7 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) { return pRes->tsrow; } -static UNUSED_FUNC bool tscHashRemainDataInSubqueryResultSet(SSqlObj *pSql) { +static bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) { bool hasData = true; SSqlCmd *pCmd = &pSql->cmd; @@ -2045,8 +2148,7 @@ static UNUSED_FUNC bool tscHashRemainDataInSubqueryResultSet(SSqlObj *pSql) { SQueryInfo *pQueryInfo1 = tscGetQueryInfoDetail(&pSql->pSubs[i]->cmd, 0); if ((pRes1->row >= pRes1->numOfRows && tscHasReachLimitation(pQueryInfo1, pRes1) && - tscProjectionQueryOnTable(pQueryInfo1)) || - (pRes1->numOfRows == 0)) { + tscIsProjectionQuery(pQueryInfo1)) || (pRes1->numOfRows == 0)) { hasData = false; break; } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index b1f5a99b67..01f2bb7dd5 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -22,7 +22,7 @@ #include "tkey.h" #include "tmd5.h" #include "tscProfile.h" -#include "tscSecondaryMerge.h" +#include "tscLocalMerge.h" #include "tscSubquery.h" #include "tschemautil.h" #include "tsclient.h" @@ -32,7 +32,7 @@ static void freeQueryInfoImpl(SQueryInfo* pQueryInfo); static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache); - + SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) { if (pTagCond->pCond == NULL) { return NULL; @@ -70,13 +70,6 @@ void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, SBufferWriter* bw) { taosArrayPush(pTagCond->pCond, &cond); } -bool tscQueryOnSTable(SSqlCmd* pCmd) { - SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); - - return ((pQueryInfo->type & TSDB_QUERY_TYPE_STABLE_QUERY) == TSDB_QUERY_TYPE_STABLE_QUERY) && - (pCmd->msgType == TSDB_MSG_TYPE_QUERY); -} - bool tscQueryTags(SQueryInfo* pQueryInfo) { for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) { SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i); @@ -95,32 +88,8 @@ bool tscQueryTags(SQueryInfo* pQueryInfo) { return true; } -bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd) { - bool hasTags = false; - int32_t numOfSelectivity = 0; - - SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); - - for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) { - int32_t functId = tscSqlExprGet(pQueryInfo, i)->functionId; - if (functId == TSDB_FUNC_TAG_DUMMY) { - hasTags = true; - continue; - } - - if ((aAggs[functId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0) { - numOfSelectivity++; - } - } - - if (numOfSelectivity > 0 && hasTags) { - return true; - } - - return false; -} - -void tscGetDBInfoFromMeterId(char* tableId, char* db) { +// todo refactor, extract methods and move the common module +void tscGetDBInfoFromTableFullName(char* tableId, char* db) { char* st = strstr(tableId, TS_PATH_DELIMITER); if (st != NULL) { char* end = strstr(st + 1, TS_PATH_DELIMITER); @@ -181,8 +150,14 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { for (int32_t i = 0; i < 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) { + + if (functionId != TSDB_FUNC_PRJ && + functionId != TSDB_FUNC_TAGPRJ && + functionId != TSDB_FUNC_TAG && + functionId != TSDB_FUNC_TS && + functionId != TSDB_FUNC_ARITHM && + functionId != TSDB_FUNC_TS_COMP && + functionId != TSDB_FUNC_TID_TAG) { return false; } } @@ -209,10 +184,14 @@ bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableInde return pQueryInfo->order.orderColId >= 0; } -bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo) { - for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) { +bool tscIsProjectionQuery(SQueryInfo* pQueryInfo) { + size_t size = tscSqlExprNumOfExprs(pQueryInfo); + + for (int32_t i = 0; i < size; ++i) { int32_t functionId = tscSqlExprGet(pQueryInfo, i)->functionId; - if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TS) { + + if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG && + functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) { return false; } } @@ -225,9 +204,10 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) { for (int32_t i = 0; i < size; ++i) { SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i); - if (pExpr == NULL) { - return false; - } + assert(pExpr != NULL); +// if (pExpr == NULL) { +// return false; +// } int32_t functionId = pExpr->functionId; if (functionId == TSDB_FUNC_TAG) { @@ -238,6 +218,7 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) { return false; } } + return true; } @@ -1772,7 +1753,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void SQueryInfo* pPrevQueryInfo = tscGetQueryInfoDetail(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex); pNewQueryInfo->type = pPrevQueryInfo->type; } else { - pNewQueryInfo->type |= TSDB_QUERY_TYPE_SUBQUERY; // it must be the subquery + TSDB_QUERY_SET_TYPE(pNewQueryInfo->type, TSDB_QUERY_TYPE_SUBQUERY);// it must be the subquery } uint64_t uid = pTableMetaInfo->pTableMeta->uid; @@ -1797,19 +1778,26 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void } // make sure the the sqlExpr for each fields is correct -// todo handle the agg arithmetic expression + // todo handle the agg arithmetic expression + numOfExprs = tscSqlExprNumOfExprs(pNewQueryInfo); + for(int32_t f = 0; f < pNewQueryInfo->fieldsInfo.numOfOutput; ++f) { TAOS_FIELD* field = tscFieldInfoGetField(&pNewQueryInfo->fieldsInfo, f); - numOfExprs = tscSqlExprNumOfExprs(pNewQueryInfo); - + bool matched = false; + for(int32_t k1 = 0; k1 < numOfExprs; ++k1) { SSqlExpr* pExpr1 = tscSqlExprGet(pNewQueryInfo, k1); - - if (strcmp(field->name, pExpr1->aliasName) == 0) { // eatablish link according to the result field name + + if (strcmp(field->name, pExpr1->aliasName) == 0) { // establish link according to the result field name SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pNewQueryInfo->fieldsInfo, f); pInfo->pSqlExpr = pExpr1; + + matched = true; + break; } } + + assert(matched); } tscFieldInfoUpdateOffset(pNewQueryInfo); @@ -1898,16 +1886,21 @@ void tscDoQuery(SSqlObj* pSql) { } if (QUERY_IS_JOIN_QUERY(type)) { - if ((pQueryInfo->type & TSDB_QUERY_TYPE_SUBQUERY) == 0) { + if (!TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_SUBQUERY)) { tscHandleMasterJoinQuery(pSql); - return; - } else { - // for first stage sub query, iterate all vnodes to get all timestamp - if ((pQueryInfo->type & TSDB_QUERY_TYPE_JOIN_SEC_STAGE) != TSDB_QUERY_TYPE_JOIN_SEC_STAGE) { -// doProcessSql(pSql); - assert(0); + } else { // for first stage sub query, iterate all vnodes to get all timestamp + if (!TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { + tscProcessSql(pSql); + } else { // secondary stage join query. + if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query + tscHandleMasterSTableQuery(pSql); + } else { + tscProcessSql(pSql); + } } } + + return; } else if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query tscHandleMasterSTableQuery(pSql); return; @@ -1917,11 +1910,13 @@ void tscDoQuery(SSqlObj* pSql) { } } -int16_t tscGetJoinTagColIndexByUid(STagCond* pTagCond, uint64_t uid) { +int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid) { if (pTagCond->joinInfo.left.uid == uid) { - return pTagCond->joinInfo.left.tagCol; + return pTagCond->joinInfo.left.tagColId; + } else if (pTagCond->joinInfo.right.uid == uid) { + return pTagCond->joinInfo.right.tagColId; } else { - return pTagCond->joinInfo.right.tagCol; + assert(0); } } @@ -1978,11 +1973,10 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) { return false; } - SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); - - STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); assert(pRes->completed); - + SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); + STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + // for normal table, no need to try any more if results are all retrieved from one vnode if (!UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo) || (pTableMetaInfo->vgroupList == NULL)) { return false; @@ -2004,7 +1998,6 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) { * if case of: multi-vnode super table projection query */ assert(pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && !tscHasReachLimitation(pQueryInfo, pRes)); - STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; diff --git a/src/common/src/tdataformat.c b/src/common/src/tdataformat.c index 40cdac9833..8dcbbf2c0b 100644 --- a/src/common/src/tdataformat.c +++ b/src/common/src/tdataformat.c @@ -105,7 +105,7 @@ void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version) { } int tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int16_t colId, int32_t bytes) { - if (!isValidDataType(type, 0)) return -1; + if (!isValidDataType(type)) return -1; if (pBuilder->nCols >= pBuilder->tCols) { pBuilder->tCols *= 2; diff --git a/src/common/src/ttypes.c b/src/common/src/ttypes.c index 4aee027c0b..3c63e2deac 100644 --- a/src/common/src/ttypes.c +++ b/src/common/src/ttypes.c @@ -363,16 +363,8 @@ char tTokenTypeSwitcher[13] = { TSDB_DATA_TYPE_NCHAR, // TK_NCHAR }; -bool isValidDataType(int32_t type, int32_t length) { - if (type < TSDB_DATA_TYPE_NULL || type > TSDB_DATA_TYPE_NCHAR) { - return false; - } - - if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) { -// return length >= 0 && length <= TSDB_MAX_BINARY_LEN; - } - - return true; +bool isValidDataType(int32_t type) { + return type >= TSDB_DATA_TYPE_NULL && type <= TSDB_DATA_TYPE_NCHAR; } bool isNull(const char *val, int32_t type) { diff --git a/src/inc/query.h b/src/inc/query.h index 49ee5248f5..5fd2ede034 100644 --- a/src/inc/query.h +++ b/src/inc/query.h @@ -32,11 +32,11 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryTableMs /** * Destroy QInfo object - * - * @param qinfo - * @return + * @param qinfo qhandle + * @param fp destroy callback function, while the qhandle is destoried, invoke the fp + * @param param free callback params */ -void qDestroyQueryInfo(qinfo_t qinfo); +void qDestroyQueryInfo(qinfo_t qinfo, void (*fp)(void*), void* param); /** * the main query execution function, including query on both table and multitables, @@ -45,7 +45,7 @@ void qDestroyQueryInfo(qinfo_t qinfo); * @param qinfo * @return */ -void qTableQuery(qinfo_t qinfo); +void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param); /** * Retrieve the produced results information, if current query is not paused or completed, @@ -80,9 +80,12 @@ bool qHasMoreResultsToRetrieve(qinfo_t qinfo); /** * kill current ongoing query and free query handle automatically - * @param qinfo + * @param qinfo qhandle + * @param fp destroy callback function, while the qhandle is destoried, invoke the fp + * @param param free callback params + * @return */ -int32_t qKillQuery(qinfo_t qinfo); +int32_t qKillQuery(qinfo_t qinfo, void (*fp)(void*), void* param); #ifdef __cplusplus } diff --git a/src/inc/taosdef.h b/src/inc/taosdef.h index 7490de90d0..8b82e1f149 100644 --- a/src/inc/taosdef.h +++ b/src/inc/taosdef.h @@ -156,7 +156,7 @@ typedef struct tDataTypeDescriptor { extern tDataTypeDescriptor tDataTypeDesc[11]; #define POINTER_BYTES sizeof(void *) // 8 by default assert(sizeof(ptrdiff_t) == sizseof(void*) -bool isValidDataType(int32_t type, int32_t length); +bool isValidDataType(int32_t type); bool isNull(const char *val, int32_t type); void setVardataNull(char* val, int32_t type); diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index 50b7527675..ba95598646 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -200,6 +200,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_MSG, 0, 0x0701, "query inva TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NO_DISKSPACE, 0, 0x0702, "query no diskspace") TAOS_DEFINE_ERROR(TSDB_CODE_QRY_OUT_OF_MEMORY, 0, 0x0703, "query out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_QRY_APP_ERROR, 0, 0x0704, "query app error") +TAOS_DEFINE_ERROR(TSDB_CODE_QRY_DUP_JOIN_KEY, 0, 0x0705, "query duplicated join key") // grant TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED, 0, 0x0800, "grant expired") diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h index 43e34b6ce2..fcb6d66422 100644 --- a/src/inc/taosmsg.h +++ b/src/inc/taosmsg.h @@ -455,7 +455,7 @@ typedef struct { int16_t orderType; // used in group by xx order by xxx int64_t limit; int64_t offset; - uint16_t queryType; // denote another query process + uint32_t queryType; // denote another query process int16_t numOfOutput; // final output columns numbers int16_t tagNameRelType; // relation of tag criteria and tbname criteria int16_t fillType; // interpolate type diff --git a/src/query/inc/qtsbuf.h b/src/query/inc/qtsbuf.h index c83c3dbe25..e437e1c4e5 100644 --- a/src/query/inc/qtsbuf.h +++ b/src/query/inc/qtsbuf.h @@ -100,10 +100,10 @@ typedef struct STSBuf { typedef struct STSBufFileHeader { uint32_t magic; // file magic number uint32_t numOfVnode; // number of vnode stored in current file - uint32_t tsOrder; // timestamp order in current file + int32_t tsOrder; // timestamp order in current file } STSBufFileHeader; -STSBuf* tsBufCreate(bool autoDelete); +STSBuf* tsBufCreate(bool autoDelete, int32_t order); STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete); STSBuf* tsBufCreateFromCompBlocks(const char* pData, int32_t numOfBlocks, int32_t len, int32_t tsOrder); diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index 618051aab2..0728fcaa35 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -12,6 +12,7 @@ * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ +#include #include "os.h" #include "qfill.h" @@ -1089,7 +1090,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS // from top to bottom in desc // from bottom to top in asc order if (pRuntimeEnv->pTSBuf != NULL) { - SQInfo *pQInfo = (SQInfo *)GET_QINFO_ADDR(pQuery); + SQInfo *pQInfo = (SQInfo *)GET_QINFO_ADDR(pRuntimeEnv); qTrace("QInfo:%p process data rows, numOfRows:%d, query order:%d, ts comp order:%d", pQInfo, pDataBlockInfo->rows, pQuery->order.order, pRuntimeEnv->pTSBuf->cur.order); } @@ -1399,7 +1400,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order pCtx->inputType = pQuery->colList[index].type; } - assert(isValidDataType(pCtx->inputType, pCtx->inputBytes)); + assert(isValidDataType(pCtx->inputType)); pCtx->ptsOutputBuf = NULL; pCtx->outputBytes = pQuery->pSelectExpr[i].bytes; @@ -2188,8 +2189,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { * set tag value in SQLFunctionCtx * e.g.,tag information into input buffer */ -static void doSetTagValueInParam(void *tsdb, STableId* pTableId, int32_t tagColId, tVariant *tag, int16_t type, - int16_t bytes) { +static void doSetTagValueInParam(void *tsdb, STableId* pTableId, int32_t tagColId, tVariant *tag, int16_t type, int16_t bytes) { tVariantDestroy(tag); if (tagColId == TSDB_TBNAME_COLUMN_INDEX) { @@ -2205,8 +2205,18 @@ static void doSetTagValueInParam(void *tsdb, STableId* pTableId, int32_t tagColI } if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) { + if (isNull(varDataVal(val), type)) { + tag->nType = TSDB_DATA_TYPE_NULL; + return; + } + tVariantCreateFromBinary(tag, varDataVal(val), varDataLen(val), type); } else { + if (isNull(val, type)) { + tag->nType = TSDB_DATA_TYPE_NULL; + return; + } + tVariantCreateFromBinary(tag, val, bytes, type); } } @@ -2214,35 +2224,55 @@ static void doSetTagValueInParam(void *tsdb, STableId* pTableId, int32_t tagColI void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, STableId* pTableId, void *tsdb) { SQuery *pQuery = pRuntimeEnv->pQuery; + SQInfo* pQInfo = GET_QINFO_ADDR(pRuntimeEnv); SExprInfo *pExprInfo = &pQuery->pSelectExpr[0]; if (pQuery->numOfOutput == 1 && pExprInfo->base.functionId == TSDB_FUNC_TS_COMP) { - assert(pExprInfo->base.numOfParams == 1); - doSetTagValueInParam(tsdb, pTableId, pExprInfo->base.arg->argValue.i64, &pRuntimeEnv->pCtx[0].tag, - pExprInfo->type, pExprInfo->bytes); + + // todo refactor extract function. + int16_t type = -1, bytes = -1; + for(int32_t i = 0; i < pQuery->numOfTags; ++i) { + if (pQuery->tagColList[i].colId == pExprInfo->base.arg->argValue.i64) { + type = pQuery->tagColList[i].type; + bytes = pQuery->tagColList[i].bytes; + } + } + + doSetTagValueInParam(tsdb, pTableId, pExprInfo->base.arg->argValue.i64, &pRuntimeEnv->pCtx[0].tag, type, bytes); } else { // set tag value, by which the results are aggregated. for (int32_t idx = 0; idx < pQuery->numOfOutput; ++idx) { - SExprInfo* pExprInfo = &pQuery->pSelectExpr[idx]; + SExprInfo* pLocalExprInfo = &pQuery->pSelectExpr[idx]; // ts_comp column required the tag value for join filter - if (!TSDB_COL_IS_TAG(pExprInfo->base.colInfo.flag)) { + if (!TSDB_COL_IS_TAG(pLocalExprInfo->base.colInfo.flag)) { continue; } // todo use tag column index to optimize performance - doSetTagValueInParam(tsdb, pTableId, pExprInfo->base.colInfo.colId, &pRuntimeEnv->pCtx[idx].tag, - pExprInfo->type, pExprInfo->bytes); + doSetTagValueInParam(tsdb, pTableId, pLocalExprInfo->base.colInfo.colId, &pRuntimeEnv->pCtx[idx].tag, + pLocalExprInfo->type, pLocalExprInfo->bytes); } // set the join tag for first column SSqlFuncMsg *pFuncMsg = &pExprInfo->base; - if (pFuncMsg->functionId == TSDB_FUNC_TS && pFuncMsg->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX && + if ((pFuncMsg->functionId == TSDB_FUNC_TS || pFuncMsg->functionId == TSDB_FUNC_PRJ) && pFuncMsg->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX && pRuntimeEnv->pTSBuf != NULL) { assert(pFuncMsg->numOfParams == 1); - assert(0); // to do fix me - // doSetTagValueInParam(pTagSchema, pFuncMsg->arg->argValue.i64, pMeterSidInfo, &pRuntimeEnv->pCtx[0].tag); + + // todo refactor + int16_t type = -1, bytes = -1; + for(int32_t i = 0; i < pQuery->numOfTags; ++i) { + if (pQuery->tagColList[i].colId == pExprInfo->base.arg->argValue.i64) { + type = pQuery->tagColList[i].type; + bytes = pQuery->tagColList[i].bytes; + } + } + + doSetTagValueInParam(tsdb, pTableId, pExprInfo->base.arg->argValue.i64, &pRuntimeEnv->pCtx[0].tag, type, bytes); + qTrace("QInfo:%p set tag value for join comparison, colId:%d, val:%"PRId64, pQInfo, pExprInfo->base.arg->argValue.i64, + pRuntimeEnv->pCtx[0].tag) } } } @@ -3617,6 +3647,7 @@ bool queryHasRemainResults(SQueryRuntimeEnv* pRuntimeEnv) { static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data) { SQuery *pQuery = pQInfo->runtimeEnv.pQuery; + for (int32_t col = 0; col < pQuery->numOfOutput; ++col) { int32_t bytes = pQuery->pSelectExpr[col].bytes; @@ -4263,12 +4294,12 @@ static void sequentialTableProcess(SQInfo *pQInfo) { assert(taosArrayGetSize(s) >= 1); setTagVal(pRuntimeEnv, (STableId*) taosArrayGet(s, 0), pQInfo->tsdb); - if (isFirstLastRowQuery(pQuery)) { assert(taosArrayGetSize(s) == 1); } + taosArrayDestroy(s); - + // here we simply set the first table as current table pQuery->current = ((SGroupItem*) taosArrayGet(group, 0))->info; scanOneTableDataBlocks(pRuntimeEnv, pQuery->current->lastKey); @@ -4454,6 +4485,10 @@ static void sequentialTableProcess(SQInfo *pQInfo) { } } } + + if (pQInfo->tableIndex >= pQInfo->groupInfo.numOfTables) { + setQueryStatus(pQuery, QUERY_COMPLETED); + } } /* @@ -5021,7 +5056,7 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, pQueryMsg->order = htons(pQueryMsg->order); pQueryMsg->orderColId = htons(pQueryMsg->orderColId); - pQueryMsg->queryType = htons(pQueryMsg->queryType); + pQueryMsg->queryType = htonl(pQueryMsg->queryType); pQueryMsg->tagNameRelType = htons(pQueryMsg->tagNameRelType); pQueryMsg->numOfCols = htons(pQueryMsg->numOfCols); @@ -5040,7 +5075,6 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, } char *pMsg = (char *)(pQueryMsg->colList) + sizeof(SColumnInfo) * pQueryMsg->numOfCols; - for (int32_t col = 0; col < pQueryMsg->numOfCols; ++col) { SColumnInfo *pColInfo = &pQueryMsg->colList[col]; @@ -5190,11 +5224,11 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, pMsg += len; } - qTrace("qmsg:%p query %d tables, qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, order:%d, " - "outputCols:%d, numOfCols:%d, interval:%" PRId64 ", fillType:%d, comptsLen:%d, limit:%" PRId64 ", offset:%" PRId64, - pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols, + qTrace("qmsg:%p query %d tables, type:%d, qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, order:%d, " + "outputCols:%d, numOfCols:%d, interval:%" PRId64 ", fillType:%d, comptsLen:%d, compNumOfBlocks:%d, limit:%" PRId64 ", offset:%" PRId64, + pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->queryType, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols, pQueryMsg->order, pQueryMsg->numOfOutput, pQueryMsg->numOfCols, pQueryMsg->intervalTime, - pQueryMsg->fillType, pQueryMsg->tsLen, pQueryMsg->limit, pQueryMsg->offset); + pQueryMsg->fillType, pQueryMsg->tsLen, pQueryMsg->tsNumOfBlocks, pQueryMsg->limit, pQueryMsg->offset); return 0; } @@ -5280,7 +5314,7 @@ static int32_t createQFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SExprInfo * if (pExprs[i].base.functionId == TSDB_FUNC_TAG_DUMMY || pExprs[i].base.functionId == TSDB_FUNC_TS_DUMMY) { tagLen += pExprs[i].bytes; } - assert(isValidDataType(pExprs[i].type, pExprs[i].bytes)); + assert(isValidDataType(pExprs[i].type)); } // TODO refactor @@ -5642,7 +5676,7 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ STSBuf *pTSBuf = NULL; if (pQueryMsg->tsLen > 0) { // open new file to save the result - char *tsBlock = (char *)pQueryMsg + pQueryMsg->tsOffset; + char *tsBlock = (char *) pQueryMsg + pQueryMsg->tsOffset; pTSBuf = tsBufCreateFromCompBlocks(tsBlock, pQueryMsg->tsNumOfBlocks, pQueryMsg->tsLen, pQueryMsg->tsOrder); tsBufResetPos(pTSBuf); @@ -5873,23 +5907,20 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi bool isSTableQuery = false; STableGroupInfo groupInfo = {0}; - //todo multitable_query?? - if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY|TSDB_QUERY_TYPE_TABLE_QUERY)) { - isSTableQuery = TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY); - + if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_TABLE_QUERY)) { STableIdInfo *id = taosArrayGet(pTableIdList, 0); - qTrace("qmsg:%p query table, uid:%"PRId64", tid:%d", pQueryMsg, id->uid, id->tid); - + + qTrace("qmsg:%p query normal table, uid:%"PRId64", tid:%d", pQueryMsg, id->uid, id->tid); if ((code = tsdbGetOneTableGroup(tsdb, id->uid, &groupInfo)) != TSDB_CODE_SUCCESS) { goto _over; } - } else if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_STABLE_QUERY)) { + } else if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY|TSDB_QUERY_TYPE_STABLE_QUERY)) { isSTableQuery = true; - // TODO: need a macro from TSDB to check if table is super table, - // also note there's possiblity that only one table in the super table - if (taosArrayGetSize(pTableIdList) == 1) { + // TODO: need a macro from TSDB to check if table is super table + + // also note there's possibility that only one table in the super table + if (!TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY)) { STableIdInfo *id = taosArrayGet(pTableIdList, 0); - // if array size is 1 and assert super table // group by normal column, do not pass the group by condition to tsdb to group table into different group int32_t numOfGroupByCols = pQueryMsg->numOfGroupCols; @@ -5903,15 +5934,13 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi goto _over; } } else { - SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES); + groupInfo.pGroupList = taosArrayInit(1, POINTER_BYTES); + groupInfo.numOfTables = taosArrayGetSize(pTableIdList); - SArray* sa = taosArrayInit(groupInfo.numOfTables, sizeof(STableId)); - for(int32_t i = 0; i < groupInfo.numOfTables; ++i) { - STableIdInfo* tableId = taosArrayGet(pTableIdList, i); - taosArrayPush(sa, tableId); - } - taosArrayPush(pTableGroup, &sa); - groupInfo.pGroupList = pTableGroup; + SArray* p = taosArrayClone(pTableIdList); + taosArrayPush(groupInfo.pGroupList, &p); + + qTrace("qmsg:%p query on %d tables in one group from client", pQueryMsg, groupInfo.numOfTables); } } else { assert(0); @@ -5952,7 +5981,7 @@ static void doDestoryQueryInfo(SQInfo* pQInfo) { freeQInfo(pQInfo); } -void qDestroyQueryInfo(qinfo_t qHandle) { +void qDestroyQueryInfo(qinfo_t qHandle, void (*fp)(void*), void* param) { SQInfo* pQInfo = (SQInfo*) qHandle; if (!isValidQInfo(pQInfo)) { return; @@ -5963,10 +5992,14 @@ void qDestroyQueryInfo(qinfo_t qHandle) { if (ref == 0) { doDestoryQueryInfo(pQInfo); + + if (fp != NULL) { + fp(param); + } } } -void qTableQuery(qinfo_t qinfo) { +void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) { SQInfo *pQInfo = (SQInfo *)qinfo; if (pQInfo == NULL || pQInfo->signature != pQInfo) { @@ -5976,7 +6009,7 @@ void qTableQuery(qinfo_t qinfo) { if (isQueryKilled(pQInfo)) { qTrace("QInfo:%p it is already killed, abort", pQInfo); - qDestroyQueryInfo(pQInfo); + qDestroyQueryInfo(pQInfo, fp, param); return; } @@ -5992,7 +6025,7 @@ void qTableQuery(qinfo_t qinfo) { } sem_post(&pQInfo->dataReady); - qDestroyQueryInfo(pQInfo); + qDestroyQueryInfo(pQInfo, fp, param); } int32_t qRetrieveQueryResultInfo(qinfo_t qinfo) { @@ -6085,7 +6118,7 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co return code; } -int32_t qKillQuery(qinfo_t qinfo) { +int32_t qKillQuery(qinfo_t qinfo, void (*fp)(void*), void* param) { SQInfo *pQInfo = (SQInfo *)qinfo; if (pQInfo == NULL || !isValidQInfo(pQInfo)) { @@ -6093,7 +6126,7 @@ int32_t qKillQuery(qinfo_t qinfo) { } setQueryKilled(pQInfo); - qDestroyQueryInfo(pQInfo); + qDestroyQueryInfo(pQInfo, fp, param); return TSDB_CODE_SUCCESS; } @@ -6123,6 +6156,17 @@ static void buildTagQueryResult(SQInfo* pQInfo) { int32_t rsize = pExprInfo->bytes; count = 0; + int16_t bytes = pExprInfo->bytes; + int16_t type = pExprInfo->type; + + for(int32_t i = 0; i < pQuery->numOfTags; ++i) { + if (pQuery->tagColList[i].colId == pExprInfo->base.colInfo.colId) { + bytes = pQuery->tagColList[i].bytes; + type = pQuery->tagColList[i].type; + break; + } + } + while(pQInfo->tableIndex < num && count < pQuery->rec.capacity) { int32_t i = pQInfo->tableIndex++; SGroupItem *item = taosArrayGet(pa, i); @@ -6140,9 +6184,6 @@ static void buildTagQueryResult(SQInfo* pQInfo) { *(int32_t *)output = pQInfo->vgId; output += sizeof(pQInfo->vgId); - int16_t bytes = pExprInfo->bytes; - int16_t type = pExprInfo->type; - if (pExprInfo->base.colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) { char *data = tsdbGetTableName(pQInfo->tsdb, &item->id); memcpy(output, data, varDataTLen(data)); @@ -6159,7 +6200,7 @@ static void buildTagQueryResult(SQInfo* pQInfo) { } else { if (val == NULL) { setNull(output, type, bytes); - } else { + } else { // todo here stop will cause client crash memcpy(output, val, bytes); } } diff --git a/src/query/src/qtsbuf.c b/src/query/src/qtsbuf.c index 834f648b7a..85aa2e7903 100644 --- a/src/query/src/qtsbuf.c +++ b/src/query/src/qtsbuf.c @@ -15,7 +15,7 @@ static int32_t STSBufUpdateHeader(STSBuf* pTSBuf, STSBufFileHeader* pHeader); * @param path * @return */ -STSBuf* tsBufCreate(bool autoDelete) { +STSBuf* tsBufCreate(bool autoDelete, int32_t order) { STSBuf* pTSBuf = calloc(1, sizeof(STSBuf)); if (pTSBuf == NULL) { return NULL; @@ -40,7 +40,7 @@ STSBuf* tsBufCreate(bool autoDelete) { pTSBuf->cur.order = TSDB_ORDER_ASC; pTSBuf->autoDelete = autoDelete; - pTSBuf->tsOrder = -1; + pTSBuf->tsOrder = order; return pTSBuf; } @@ -66,8 +66,8 @@ STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete) { // validate the file magic number STSBufFileHeader header = {0}; fseek(pTSBuf->f, 0, SEEK_SET); - fread(&header, 1, sizeof(header), pTSBuf->f); - + fread(&header, 1, sizeof(STSBufFileHeader), pTSBuf->f); + // invalid file if (header.magic != TS_COMP_FILE_MAGIC) { return NULL; @@ -119,7 +119,6 @@ STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete) { // ascending by default pTSBuf->cur.order = TSDB_ORDER_ASC; - pTSBuf->autoDelete = autoDelete; // tscTrace("create tsBuf from file:%s, fd:%d, size:%d, numOfVnode:%d, autoDelete:%d", pTSBuf->path, fileno(pTSBuf->f), @@ -536,7 +535,9 @@ int32_t STSBufUpdateHeader(STSBuf* pTSBuf, STSBufFileHeader* pHeader) { if ((pTSBuf->f == NULL) || pHeader == NULL || pHeader->numOfVnode == 0 || pHeader->magic != TS_COMP_FILE_MAGIC) { return -1; } - + + assert(pHeader->tsOrder == TSDB_ORDER_ASC || pHeader->tsOrder == TSDB_ORDER_DESC); + int64_t r = fseek(pTSBuf->f, 0, SEEK_SET); if (r != 0) { return -1; @@ -754,7 +755,7 @@ int32_t tsBufMerge(STSBuf* pDestBuf, const STSBuf* pSrcBuf, int32_t vnodeId) { } STSBuf* tsBufCreateFromCompBlocks(const char* pData, int32_t numOfBlocks, int32_t len, int32_t order) { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, order); STSVnodeBlockInfo* pBlockInfo = &(addOneVnodeInfo(pTSBuf, 0)->info); pBlockInfo->numOfBlocks = numOfBlocks; @@ -845,7 +846,9 @@ STSBuf* tsBufClone(STSBuf* pTSBuf) { if (pTSBuf == NULL) { return NULL; } - + + tsBufFlush(pTSBuf); + return tsBufCreateFromFile(pTSBuf->path, false); } diff --git a/src/query/src/tvariant.c b/src/query/src/tvariant.c index 7ba4e7899e..1650db6cfd 100644 --- a/src/query/src/tvariant.c +++ b/src/query/src/tvariant.c @@ -474,6 +474,7 @@ static FORCE_INLINE int32_t convertToInteger(tVariant *pVariant, int64_t *result free(pVariant->pz); pVariant->nLen = 0; } + setNull((char *)result, type, tDataTypeDesc[type].nSize); return 0; } @@ -597,10 +598,10 @@ static int32_t convertToBool(tVariant *pVariant, int64_t *pDest) { * todo handle the return value */ int32_t tVariantDump(tVariant *pVariant, char *payload, int16_t type, bool includeLengthPrefix) { - if (pVariant == NULL || (pVariant->nType != 0 && !isValidDataType(pVariant->nType, pVariant->nLen))) { + if (pVariant == NULL || (pVariant->nType != 0 && !isValidDataType(pVariant->nType))) { return -1; } - + errno = 0; // reset global error code switch (type) { diff --git a/src/query/tests/tsBufTest.cpp b/src/query/tests/tsBufTest.cpp index f50e1a5a71..1d2d94f3c0 100644 --- a/src/query/tests/tsBufTest.cpp +++ b/src/query/tests/tsBufTest.cpp @@ -29,7 +29,7 @@ int64_t* createTsList(int32_t num, int64_t start, int32_t step) { // simple test void simpleTest() { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); // write 10 ts points int32_t num = 10; @@ -52,7 +52,7 @@ void simpleTest() { // one large list of ts, the ts list need to be split into several small blocks void largeTSTest() { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); // write 10 ts points int32_t num = 1000000; @@ -75,7 +75,7 @@ void largeTSTest() { } void multiTagsTest() { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); int32_t num = 10000; int64_t tag = 1; @@ -105,7 +105,7 @@ void multiTagsTest() { } void multiVnodeTagsTest() { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); int32_t num = 10000; int64_t start = 10000000; @@ -143,7 +143,7 @@ void multiVnodeTagsTest() { } void loadDataTest() { - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); int32_t num = 10000; int64_t oldStart = 10000000; @@ -221,7 +221,7 @@ void TSTraverse() { int32_t step = 30; int32_t numOfVnode = 2; - STSBuf* pTSBuf = tsBufCreate(true); + STSBuf* pTSBuf = tsBufCreate(true, TSDB_ORDER_ASC); for (int32_t j = 0; j < numOfVnode; ++j) { // vnodeId:0 @@ -359,8 +359,8 @@ void invalidFileTest() { } void mergeDiffVnodeBufferTest() { - STSBuf* pTSBuf1 = tsBufCreate(true); - STSBuf* pTSBuf2 = tsBufCreate(true); + STSBuf* pTSBuf1 = tsBufCreate(true, TSDB_ORDER_ASC); + STSBuf* pTSBuf2 = tsBufCreate(true, TSDB_ORDER_ASC); int32_t step = 30; int32_t num = 1000; @@ -391,8 +391,8 @@ void mergeDiffVnodeBufferTest() { } void mergeIdenticalVnodeBufferTest() { - STSBuf* pTSBuf1 = tsBufCreate(true); - STSBuf* pTSBuf2 = tsBufCreate(true); + STSBuf* pTSBuf1 = tsBufCreate(true, TSDB_ORDER_ASC); + STSBuf* pTSBuf2 = tsBufCreate(true, TSDB_ORDER_ASC); int32_t step = 30; int32_t num = 1000; diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index a62ad5bbd3..732e8e9008 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -1930,8 +1930,7 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) { SColIndex* pColIndex = &pTableGroupSupp->pCols[i]; int32_t colIndex = pColIndex->colIndex; - assert((colIndex >= 0 && colIndex < schemaNCols(pTableGroupSupp->pTagSchema)) || - (colIndex == TSDB_TBNAME_COLUMN_INDEX)); + assert(colIndex >= TSDB_TBNAME_COLUMN_INDEX); char * f1 = NULL; char * f2 = NULL; @@ -1950,7 +1949,20 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) { f1 = tdGetKVRowValOfCol(pTable1->tagVal, pCol->colId); f2 = tdGetKVRowValOfCol(pTable2->tagVal, pCol->colId); } - + + // this tags value may be NULL + if (f1 == NULL && f2 == NULL) { + continue; + } + + if (f1 == NULL) { + return -1; + } + + if (f2 == NULL) { + return 1; + } + int32_t ret = doCompare(f1, f2, type, bytes); if (ret == 0) { continue; @@ -2112,8 +2124,8 @@ int32_t tsdbQuerySTableByTagCond(TsdbRepoT* tsdb, uint64_t uid, const char* pTag } if (pTable->type != TSDB_SUPER_TABLE) { - tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", tid:%d, name:%s", - tsdb, uid, pTable->tableId.tid, pTable->name); + tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", tid:%d, name:%s", tsdb, uid, pTable->tableId.tid, + pTable->name); return TSDB_CODE_COM_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client } @@ -2128,7 +2140,7 @@ int32_t tsdbQuerySTableByTagCond(TsdbRepoT* tsdb, uint64_t uid, const char* pTag pGroupInfo->numOfTables = taosArrayGetSize(res); pGroupInfo->pGroupList = createTableGroup(res, pTagSchema, pColIndex, numOfCols, tsdb); - tsdbTrace("no tbname condition or tagcond, all tables belongs to one group, numOfTables:%d", pGroupInfo->numOfTables); + tsdbTrace("%p no table name/tag condition, all tables belong to one group, numOfTables:%d", tsdb, pGroupInfo->numOfTables); } else { // todo add error } @@ -2172,6 +2184,9 @@ int32_t tsdbQuerySTableByTagCond(TsdbRepoT* tsdb, uint64_t uid, const char* pTag pGroupInfo->numOfTables = taosArrayGetSize(res); pGroupInfo->pGroupList = createTableGroup(res, pTagSchema, pColIndex, numOfCols, tsdb); + tsdbTrace("%p stable tid:%d, uid:%"PRIu64" query, numOfTables:%d, belong to %d groups", tsdb, pTable->tableId.tid, + pTable->tableId.uid, pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList)); + taosArrayDestroy(res); return ret; } diff --git a/src/util/inc/tutil.h b/src/util/inc/tutil.h index d38f983718..c75ba24baa 100644 --- a/src/util/inc/tutil.h +++ b/src/util/inc/tutil.h @@ -34,18 +34,19 @@ extern "C" { #define WCHAR wchar_t -#define tfree(x) \ - { \ - if (x) { \ - free((void*)(x)); \ - x = 0; \ - } \ +#define tfree(x) \ + { \ + if (x) { \ + free((void *)(x)); \ + x = 0; \ + } \ } -#define tstrncpy(dst, src, size) do { \ +#define tstrncpy(dst, src, size) \ + do { \ strncpy((dst), (src), (size)); \ - (dst)[(size) - 1] = 0; \ -} while (0); + (dst)[(size)-1] = 0; \ + } while (0); #define tclose(x) taosCloseSocket(x) diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index 1e770d8d27..f9dcd5e6e0 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -89,7 +89,10 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { vWarn("QInfo:%p connection %p broken, kill query", killQueryMsg->qhandle, pReadMsg->rpcMsg.handle); assert(pReadMsg->rpcMsg.contLen > 0 && killQueryMsg->free == 1); - qKillQuery((qinfo_t) killQueryMsg->qhandle); + // this message arrived here by means of the query message, so release the vnode is necessary + qKillQuery((qinfo_t) killQueryMsg->qhandle, vnodeRelease, pVnode); + vnodeRelease(pVnode); + return TSDB_CODE_TSC_QUERY_CANCELLED; // todo change the error code } @@ -112,8 +115,8 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; //NOTE: there two refcount, needs to kill twice, todo refactor - qKillQuery(pQInfo); - qKillQuery(pQInfo); + qKillQuery(pQInfo, vnodeRelease, pVnode); + qKillQuery(pQInfo, vnodeRelease, pVnode); return pRsp->code; } @@ -128,7 +131,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { if (pQInfo != NULL) { vTrace("vgId:%d, QInfo:%p, do qTableQuery", pVnode->vgId, pQInfo); - qTableQuery(pQInfo); // do execute query + qTableQuery(pQInfo, vnodeRelease, pVnode); // do execute query } return code; @@ -146,7 +149,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { if (pRetrieve->free == 1) { vTrace("vgId:%d, QInfo:%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, pQInfo); - int32_t ret = qKillQuery(pQInfo); + int32_t ret = qKillQuery(pQInfo, vnodeRelease, pVnode); pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); pRet->len = sizeof(SRetrieveTableRsp); @@ -175,8 +178,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { pRet->qhandle = pQInfo; code = TSDB_CODE_VND_ACTION_NEED_REPROCESSED; } else { // no further execution invoked, release the ref to vnode - qDestroyQueryInfo(pQInfo); - vnodeRelease(pVnode); + qDestroyQueryInfo(pQInfo, vnodeRelease, pVnode); } } diff --git a/tests/examples/c/demo.c b/tests/examples/c/demo.c index 34c785a271..5030af4ad6 100644 --- a/tests/examples/c/demo.c +++ b/tests/examples/c/demo.c @@ -13,145 +13,55 @@ * along with this program. If not, see . */ -// TAOS standard API example. The same syntax as MySQL, but only a subet +// TAOS standard API example. The same syntax as MySQL, but only a subset // to compile: gcc -o demo demo.c -ltaos -#include #include #include #include #include // TAOS header file -#include -#include - -static int32_t doQuery(TAOS* taos, const char* sql) { - struct timeval t1 = {0}; - gettimeofday(&t1, NULL); - - TAOS_RES* res = taos_query(taos, sql); - if (taos_errno(res) != 0) { - printf("failed to execute query, reason:%s\n", taos_errstr(taos)); - return -1; - } - - TAOS_ROW row = NULL; - char buf[512] = {0}; - - int32_t numOfFields = taos_num_fields(res); - TAOS_FIELD* pFields = taos_fetch_fields(res); - - int32_t i = 0; - while((row = taos_fetch_row(res)) != NULL) { - taos_print_row(buf, row, pFields, numOfFields); - printf("%d:%s\n", ++i, buf); - memset(buf, 0, 512); - } - - taos_free_result(res); - - struct timeval t2 = {0}; - gettimeofday(&t2, NULL); - - printf("elapsed time:%"PRId64 " ms\n", ((t2.tv_sec*1000000 + t2.tv_usec) - (t1.tv_sec*1000000 + t1.tv_usec))/1000); - return 0; -} - -void* oneLoader(void* param) { - TAOS* conn = (TAOS*) param; - - for(int32_t i = 0; i < 20000; ++i) { -// doQuery(conn, "show databases"); - doQuery(conn, "use test"); -// doQuery(conn, "describe t12"); -// doQuery(conn, "show tables"); -// doQuery(conn, "create table if not exists abc (ts timestamp, k int)"); -// doQuery(conn, "select * from t12"); - } - - return 0; -} - - -static __attribute__((unused)) void multiThreadTest(int32_t numOfThreads, void* conn) { - pthread_attr_t thattr; - pthread_attr_init(&thattr); - pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE); - - pthread_t* threadId = (pthread_t*)malloc(sizeof(pthread_t)*(uint32_t)numOfThreads); - - for (int i = 0; i < numOfThreads; ++i) { - pthread_create(&threadId[i], NULL, oneLoader, conn); - } - - for (int32_t i = 0; i < numOfThreads; ++i) { - pthread_join(threadId[i], NULL); - } - - free(threadId); - pthread_attr_destroy(&thattr); -} int main(int argc, char *argv[]) { TAOS * taos; char qstr[1024]; TAOS_RES *result; - + // connect to server if (argc < 2) { printf("please input server-ip \n"); return 0; } - - taos_options(TSDB_OPTION_CONFIGDIR, "~/sec/cfg"); - + // init TAOS taos_init(); - + taos = taos_connect(argv[1], "root", "taosdata", NULL, 0); if (taos == NULL) { - printf("failed to connect to server, reason:%s\n", taos_errstr(NULL)); + printf("failed to connect to server, reason:%s\n", taos_errstr(taos)); exit(1); } - printf("success to connect to server\n"); -// doQuery(taos, "select c1,count(*) from group_db0.group_mt0 where c1<8 group by c1"); -// doQuery(taos, "select * from test.m1"); -// multiThreadTest(1, taos); -// doQuery(taos, "select tbname from test.m1"); -// doQuery(taos, "select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from lm2_db0.lm2_stb0 where ts >= 1537146000000 and ts <= 1543145400000 and tbname in ('lm2_tb0') interval(1s) group by t1"); -// doQuery(taos, "select max(c1), min(c2), sum(c3), avg(c4), first(c7), last(c8), first(c9) from lm2_db0.lm2_stb0 where ts >= 1537146000000 and ts <= 1543145400000 and tbname in ('lm2_tb0', 'lm2_tb1', 'lm2_tb2') interval(1s)"); - for(int32_t i = 0; i < 200; ++i) { - doQuery(taos, "select * from lm2_db0.lm2_stb0"); - } -// doQuery(taos, "create table t1(ts timestamp, k binary(12), f nchar(2))"); - - taos_close(taos); - return 0; - + taos_query(taos, "drop database demo"); if (taos_query(taos, "create database demo") != 0) { printf("failed to create database, reason:%s\n", taos_errstr(taos)); exit(1); } printf("success to create database\n"); - - + taos_query(taos, "use demo"); - - + // create table if (taos_query(taos, "create table m1 (ts timestamp, speed int)") != 0) { printf("failed to create table, reason:%s\n", taos_errstr(taos)); exit(1); } printf("success to create table\n"); - - + // sleep for one second to make sure table is created on data node // taosMsleep(1000); - - + // insert 10 records int i = 0; for (i = 0; i < 10; ++i) { @@ -162,29 +72,21 @@ int main(int argc, char *argv[]) { //sleep(1); } printf("success to insert rows, total %d rows\n", i); - - + // query the records sprintf(qstr, "SELECT * FROM m1"); - if (taos_query(taos, qstr) != 0) { - printf("failed to select, reason:%s\n", taos_errstr(taos)); - exit(1); - } - - if (result == NULL) { - printf("failed to get result, reason:%s\n", taos_errstr(taos)); + result = taos_query(taos, qstr); + if (result == NULL || taos_errno(result) != 0) { + printf("failed to select, reason:%s\n", taos_errstr(result)); exit(1); } -// TAOS_ROW row; - TAOS_ROW row; int rows = 0; int num_fields = taos_field_count(taos); TAOS_FIELD *fields = taos_fetch_fields(result); char temp[256]; - - + printf("select * from table, result:\n"); // fetch the records row by row while ((row = taos_fetch_row(result))) { @@ -192,7 +94,7 @@ int main(int argc, char *argv[]) { taos_print_row(temp, row, fields, num_fields); printf("%s\n", temp); } - + taos_free_result(result); printf("====demo end====\n\n"); return getchar(); diff --git a/tests/script/general/parser/alter.sim b/tests/script/general/parser/alter.sim index c30cc171bc..018ee924d1 100644 --- a/tests/script/general/parser/alter.sim +++ b/tests/script/general/parser/alter.sim @@ -135,7 +135,7 @@ endi sql alter table tb1 drop column c3 sleep 6000 sql insert into tb1 values (now, 2, 'taos') -sleep 3000 +sleep 30000 sql select * from strm if $rows != 2 then return -1 diff --git a/tests/script/general/parser/alter1.sim b/tests/script/general/parser/alter1.sim index 7c4609a87d..24cf79eae1 100644 --- a/tests/script/general/parser/alter1.sim +++ b/tests/script/general/parser/alter1.sim @@ -108,11 +108,12 @@ endi print ================== change a tag value sql alter table car1 set tag carid=10 -sql describe car1 -if $rows != 7 then +sql select carId, carmodel from car1 +if $rows != 1 then return -1 endi -if $data43 != 10 then + +if $data00 != 10 then return -1 endi diff --git a/tests/script/general/parser/join.sim b/tests/script/general/parser/join.sim index 2524fbe631..1bce6f1950 100644 --- a/tests/script/general/parser/join.sim +++ b/tests/script/general/parser/join.sim @@ -258,6 +258,7 @@ sql select count(join_tb1.c3), count(join_tb0.ts) from $tb1 , $tb2 where $ts1 = $val = 2 if $data00 != $val then + print expect 2, actaul: $data00 return -1 endi @@ -353,6 +354,7 @@ sql select count(*) from join_mt0, join_mt1 where join_mt0.ts = join_mt1.ts and $val = 20 if $data00 != $val then + print expect 20, actual:$data00 return -1 endi @@ -410,7 +412,7 @@ endi #======================limit offset=================================== # tag values not int -sql_error select count(*) from join_mt0, join_mt1 where join_mt0.ts=join_mt1.ts and join_mt0.t2=join_mt1.t2; +sql_error select count(*) from join_mt0, join_mt1 where join_mt0.ts=join_mt1.ts and join_mt0.t2=join_mt1.t2; #!!!!! # tag type not identical sql_error select count(*) from join_mt0, join_mt1 where join_mt1.t2 = join_mt0.t1 and join_mt1.ts=join_mt0.ts; diff --git a/tests/script/general/parser/join_multivnode.sim b/tests/script/general/parser/join_multivnode.sim index 3e32064f3b..4cf1d36672 100644 --- a/tests/script/general/parser/join_multivnode.sim +++ b/tests/script/general/parser/join_multivnode.sim @@ -108,6 +108,7 @@ sql select join_mt0.ts,join_mt0.ts,join_mt0.t1 from join_mt0, join_mt1 where joi print $row if $row != 3000 then + print expect 3000, actual: $row return -1 endi diff --git a/tests/script/general/parser/set_tag_vals.sim b/tests/script/general/parser/set_tag_vals.sim index a684fd5406..9d7ac9678d 100644 --- a/tests/script/general/parser/set_tag_vals.sim +++ b/tests/script/general/parser/set_tag_vals.sim @@ -9,7 +9,7 @@ sql connect $dbPrefix = db $tbPrefix = tb $stbPrefix = stb -$tbNum = 1000 +$tbNum = 100 $rowNum = 100 $totalNum = $tbNum * $rowNum $ts0 = 1537146000000 @@ -26,7 +26,7 @@ $stb = $stbPrefix . $i sql drop database $db -x step1 step1: sql create database $db -print ====== create tables +print ====== create $tbNum tables sql use $db sql create table $stb (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 smallint, c6 tinyint, c7 bool, c8 binary(10), c9 nchar(10)) tags(t1 binary(9), t2 binary(8)) diff --git a/tests/script/general/parser/slimit_alter_tags.sim b/tests/script/general/parser/slimit_alter_tags.sim index 54bc868f6a..e8e81e8809 100644 --- a/tests/script/general/parser/slimit_alter_tags.sim +++ b/tests/script/general/parser/slimit_alter_tags.sim @@ -84,11 +84,13 @@ while $i < 10 sql alter table $tb set tag tg_added = $tg_added $i = $i + 1 endw -sql describe tb0 -if $rows != 7 then + +sql select t1,t2,tg_added from tb0 +if $rows != 1 then return -1 endi -if $data63 != tb0 then + +if $data02 != tb0 then return -1 endi diff --git a/tests/script/general/parser/testSuite.sim b/tests/script/general/parser/testSuite.sim index 67ef952c41..f288288fb7 100644 --- a/tests/script/general/parser/testSuite.sim +++ b/tests/script/general/parser/testSuite.sim @@ -1,45 +1,45 @@ -#run general/parser/alter.sim -#sleep 2000 -#run general/parser/alter1.sim -#sleep 2000 -#run general/parser/alter_stable.sim -#sleep 2000 -#run general/parser/auto_create_tb.sim -#sleep 2000 -#run general/parser/auto_create_tb_drop_tb.sim +run general/parser/alter.sim +sleep 2000 +run general/parser/alter1.sim +sleep 2000 +run general/parser/alter_stable.sim +sleep 2000 +run general/parser/auto_create_tb.sim +sleep 2000 +run general/parser/auto_create_tb_drop_tb.sim -#sleep 2000 -#run general/parser/col_arithmetic_operation.sim -#sleep 2000 -#run general/parser/columnValue.sim -#sleep 2000 -#run general/parser/commit.sim -#sleep 2000 -#run general/parser/create_db.sim -#sleep 2000 -#run general/parser/create_mt.sim -#sleep 2000 -#run general/parser/create_tb.sim -#sleep 2000 -#run general/parser/dbtbnameValidate.sim -#sleep 2000 -#run general/parser/import_commit1.sim -#sleep 2000 -#run general/parser/import_commit2.sim -#sleep 2000 -#run general/parser/import_commit3.sim -#sleep 2000 -#run general/parser/insert_tb.sim -#sleep 2000 -#run general/parser/first_last.sim -#sleep 2000 -##run general/parser/import_file.sim -#sleep 2000 -#run general/parser/lastrow.sim -#sleep 2000 -#run general/parser/nchar.sim -#sleep 2000 -##run general/parser/null_char.sim +sleep 2000 +run general/parser/col_arithmetic_operation.sim +sleep 2000 +run general/parser/columnValue.sim +sleep 2000 +run general/parser/commit.sim +sleep 2000 +run general/parser/create_db.sim +sleep 2000 +run general/parser/create_mt.sim +sleep 2000 +run general/parser/create_tb.sim +sleep 2000 +run general/parser/dbtbnameValidate.sim +sleep 2000 +run general/parser/import_commit1.sim +sleep 2000 +run general/parser/import_commit2.sim +sleep 2000 +run general/parser/import_commit3.sim +sleep 2000 +run general/parser/insert_tb.sim +sleep 2000 +run general/parser/first_last.sim +sleep 2000 +#run general/parser/import_file.sim +sleep 2000 +run general/parser/lastrow.sim +sleep 2000 +run general/parser/nchar.sim +sleep 2000 +#run general/parser/null_char.sim sleep 2000 run general/parser/single_row_in_tb.sim sleep 2000 @@ -80,26 +80,28 @@ sleep 2000 run general/parser/tags_dynamically_specifiy.sim sleep 2000 run general/parser/groupby.sim - sleep 2000 run general/parser/set_tag_vals.sim sleep 2000 -run general/parser/slimit_alter_tags.sim +run general/parser/slimit_alter_tags.sim # persistent failed + sleep 2000 -run general/parser/stream_on_sys.sim +run general/parser/join.sim sleep 2000 -run general/parser/stream.sim +run general/parser/join_multivnode.sim sleep 2000 #run general/parser/repeatAlter.sim sleep 2000 #run general/parser/repeatStream.sim -sleep 2000 -run general/parser/join.sim -sleep 2000 -run general/parser/join_multivnode.sim + sleep 2000 run general/parser/binary_escapeCharacter.sim sleep 2000 -#run general/parser/bug.sim \ No newline at end of file +run general/parser/bug.sim + +sleep 2000 +run general/parser/stream_on_sys.sim +sleep 2000 +run general/parser/stream.sim \ No newline at end of file