diff --git a/src/client/inc/tscLocalMerge.h b/src/client/inc/tscLocalMerge.h index c073f40546c6a5ceaeabeafd3d731fdf211402df..d38885ab2ee9734ae948e19d4816dcc8a8f73ce6 100644 --- a/src/client/inc/tscLocalMerge.h +++ b/src/client/inc/tscLocalMerge.h @@ -84,7 +84,7 @@ typedef struct SRetrieveSupport { SColumnModel * pFinalColModel; // colModel for final result SSubqueryState * pState; int32_t subqueryIndex; // index of current vnode in vnode list - SSqlObj * pParentSqlObj; + SSqlObj * pParentSql; tFilePage * localBuffer; // temp buffer, there is a buffer for each vnode to uint32_t numOfRetry; // record the number of retry times pthread_mutex_t queryMutex; diff --git a/src/client/inc/tscSubquery.h b/src/client/inc/tscSubquery.h index 82d490376aa8de18814c6cf9a1f3fbfb5be6ce75..d5833675aa1a761eb94402bfdc3fe04212d41d75 100644 --- a/src/client/inc/tscSubquery.h +++ b/src/client/inc/tscSubquery.h @@ -36,6 +36,8 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql); int32_t tscHandleMultivnodeInsert(SSqlObj *pSql); +int32_t tscHandleInsertRetry(SSqlObj* pSql); + void tscBuildResFromSubqueries(SSqlObj *pSql); void **doSetResultRowData(SSqlObj *pSql, bool finalResult); diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 471db9c99f73fbc6881c849f93d25e52261710e9..6ce94d5aa4d10688d7f92dfe2c9ccbfc8852bc9d 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -213,8 +213,7 @@ typedef struct SQueryInfo { typedef struct { int command; uint8_t msgType; - bool autoCreated; // if the table is missing, on-the-fly create it. during getmeterMeta - int8_t dataSourceType; // load data from file or not + bool autoCreated; // create table if it is not existed during retrieve table meta in mnode union { int32_t count; @@ -222,18 +221,23 @@ typedef struct { }; int32_t insertType; - int32_t clauseIndex; // index of multiple subclause query + int32_t clauseIndex; // index of multiple subclause query + + char * curSql; // current sql, resume position of sql after parsing paused int8_t parseFinished; + short numOfCols; uint32_t allocSize; char * payload; int32_t payloadLen; SQueryInfo **pQueryInfo; int32_t numOfClause; - char * curSql; // current sql, resume position of sql after parsing paused - void * pTableList; // referred table involved in sql int32_t batchSize; // for parameter ('?') binding and batch processing int32_t numOfParams; + + int8_t dataSourceType; // load data from file or not + int8_t submitSchema; // submit block is built with table schema + SHashObj *pTableList; // referred table involved in sql SArray *pDataBlocks; // SArray submit data blocks after parsing sql } SSqlCmd; diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index 0bfee46b4bfe740ed23668bfb51cc340cd4d195e..2de45bcc6e0a0be3f41f492d0a0b760c7585d3a4 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -431,6 +431,8 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { return; } + tscDebug("%p get tableMeta successfully", pSql); + if (pSql->pStream == NULL) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); @@ -446,20 +448,20 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { assert(code == TSDB_CODE_SUCCESS); } - - assert((tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0) && pTableMetaInfo->vgroupIndex >= 0 && pSql->param != NULL); + assert((tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0) && pSql->param != NULL); SRetrieveSupport *trs = (SRetrieveSupport *)pSql->param; - SSqlObj * pParObj = trs->pParentSqlObj; + SSqlObj * pParObj = trs->pParentSql; + // NOTE: the vgroupInfo for the queried super table must be existed here. assert(pParObj->signature == pParObj && trs->subqueryIndex == pTableMetaInfo->vgroupIndex && - tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0); + pTableMetaInfo->vgroupIndex >= 0 && pTableMetaInfo->vgroupList != NULL); - // NOTE: the vgroupInfo for the queried super table must be existed here. - assert(pTableMetaInfo->vgroupList != NULL); if ((code = tscProcessSql(pSql)) == TSDB_CODE_SUCCESS) { return; } + + goto _error; } else { // continue to process normal async query if (pCmd->parseFinished) { tscDebug("%p update table meta in local cache, continue to process sql and send corresponding query", pSql); @@ -472,18 +474,41 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { assert(code == TSDB_CODE_SUCCESS); } - // if failed to process sql, go to error handler - if ((code = tscProcessSql(pSql)) == TSDB_CODE_SUCCESS) { - return; + // in case of insert, redo parsing the sql string and build new submit data block for two reasons: + // 1. the table Id(tid & uid) may have been update, the submit block needs to be updated + // 2. vnode may need the schema information along with submit block to update its local table schema. + if (pCmd->command == TSDB_SQL_INSERT) { + tscDebug("%p redo parse sql string to build submit block", pSql); + + pCmd->parseFinished = false; + if ((code = tsParseSql(pSql, true)) == TSDB_CODE_SUCCESS) { + /* + * Discard previous built submit blocks, and then parse the sql string again and build up all submit blocks, + * and send the required submit block according to index value in supporter to server. + */ + pSql->fp = pSql->fetchFp; // restore the fp + if ((code = tscHandleInsertRetry(pSql)) == TSDB_CODE_SUCCESS) { + return; + } + } + + } else {// in case of other query type, continue + if ((code = tscProcessSql(pSql)) == TSDB_CODE_SUCCESS) { + return; + } } -// // todo update the submit message according to the new table meta -// // 1. table uid, 2. ip address -// code = tscSendMsgToServer(pSql); -// if (code == TSDB_CODE_SUCCESS) return; + + goto _error; } else { tscDebug("%p continue parse sql after get table meta", pSql); code = tsParseSql(pSql, false); + if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { + return; + } else if (code != TSDB_CODE_SUCCESS) { + goto _error; + } + if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_STMT_INSERT)) { STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); code = tscGetTableMeta(pSql, pTableMetaInfo); @@ -492,45 +517,49 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { } else { assert(code == TSDB_CODE_SUCCESS); } + (*pSql->fp)(pSql->param, pSql, code); return; } - - if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return; + + // proceed to invoke the tscDoQuery(); } } } else { // stream computing STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); - code = tscGetTableMeta(pSql, pTableMetaInfo); - pRes->code = code; - if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return; + code = tscGetTableMeta(pSql, pTableMetaInfo); + if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { + return; + } else if (code != TSDB_CODE_SUCCESS) { + goto _error; + } if (code == TSDB_CODE_SUCCESS && UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { code = tscGetSTableVgroupInfo(pSql, pCmd->clauseIndex); - pRes->code = code; - - if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return; + if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { + return; + } else if (code != TSDB_CODE_SUCCESS) { + goto _error; + } } - } - - if (code != TSDB_CODE_SUCCESS) { - pSql->res.code = code; - tscQueueAsyncRes(pSql); - return; - } - if (pSql->pStream) { tscDebug("%p stream:%p meta is updated, start new query, command:%d", pSql, pSql->pStream, pSql->cmd.command); if (!pSql->cmd.parseFinished) { tsParseSql(pSql, false); sem_post(&pSql->rspSem); } + return; - } else { - tscDebug("%p get tableMeta successfully", pSql); } tscDoQuery(pSql); + return; + + _error: + if (code != TSDB_CODE_SUCCESS) { + pSql->res.code = code; + tscQueueAsyncRes(pSql); + } } diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 470a82980545d839461d39f23b18bfdf3b28c8fa..8dd5ef69f7a6eb9811779debf99d77240567becd 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -347,8 +347,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) { int doProcessSql(SSqlObj *pSql) { SSqlCmd *pCmd = &pSql->cmd; SSqlRes *pRes = &pSql->res; - int32_t code = TSDB_CODE_SUCCESS; - + if (pCmd->command == TSDB_SQL_SELECT || pCmd->command == TSDB_SQL_FETCH || pCmd->command == TSDB_SQL_RETRIEVE || @@ -365,10 +364,13 @@ int doProcessSql(SSqlObj *pSql) { return pRes->code; } - code = tscSendMsgToServer(pSql); + int32_t code = tscSendMsgToServer(pSql); + + // NOTE: if code is TSDB_CODE_SUCCESS, pSql may have been released here already by other threads. if (code != TSDB_CODE_SUCCESS) { pRes->code = code; tscQueueAsyncRes(pSql); + return pRes->code; } return TSDB_CODE_SUCCESS; diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index 0c2e3f3a1c6251c828eb728095366d5aefb7ecfe..b9ba22436eb03e2507884909abca26c0281f78b9 100644 --- a/src/client/src/tscSubquery.c +++ b/src/client/src/tscSubquery.c @@ -25,6 +25,7 @@ typedef struct SInsertSupporter { SSubqueryState* pState; SSqlObj* pSql; + int32_t index; } SInsertSupporter; static void freeJoinSubqueryObj(SSqlObj* pSql); @@ -1876,13 +1877,36 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows) // release data block data tfree(pState); -// pParentCmd->pDataBlocks = tscDestroyBlockArrayList(pParentCmd->pDataBlocks); - + // restore user defined fp pParentObj->fp = pParentObj->fetchFp; - + + // todo remove this parameter in async callback function definition. // all data has been sent to vnode, call user function - (*pParentObj->fp)(pParentObj->param, pParentObj, numOfRows); + int32_t v = (pParentObj->res.code != TSDB_CODE_SUCCESS)? pParentObj->res.code:pParentObj->res.numOfRows; + (*pParentObj->fp)(pParentObj->param, pParentObj, v); +} + +/** + * it is a subquery, so after parse the sql string, copy the submit block to payload of itself + * @param pSql + * @return + */ +int32_t tscHandleInsertRetry(SSqlObj* pSql) { + assert(pSql != NULL && pSql->param != NULL); + SSqlCmd* pCmd = &pSql->cmd; + SSqlRes* pRes = &pSql->res; + + SInsertSupporter* pSupporter = (SInsertSupporter*) pSql->param; + assert(pSupporter->index < pSupporter->pState->numOfTotal); + + STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, pSupporter->index); + pRes->code = tscCopyDataBlockToPayload(pSql, pTableDataBlock); + if (pRes->code != TSDB_CODE_SUCCESS) { + return pRes->code; + } + + return tscProcessSql(pSql); } int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) { @@ -1906,10 +1930,11 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) { while(numOfSub < pSql->numOfSubs) { SInsertSupporter* pSupporter = calloc(1, sizeof(SInsertSupporter)); - pSupporter->pSql = pSql; + pSupporter->pSql = pSql; pSupporter->pState = pState; - - SSqlObj *pNew = createSimpleSubObj(pSql, multiVnodeInsertFinalize, pSupporter, TSDB_SQL_INSERT);//createSubqueryObj(pSql, 0, multiVnodeInsertFinalize, pSupporter1, TSDB_SQL_INSERT, NULL); + pSupporter->index = numOfSub; + + SSqlObj *pNew = createSimpleSubObj(pSql, multiVnodeInsertFinalize, pSupporter, TSDB_SQL_INSERT); if (pNew == NULL) { tscError("%p failed to malloc buffer for subObj, orderOfSub:%d, reason:%s", pSql, numOfSub, strerror(errno)); goto _error; @@ -1940,6 +1965,8 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) { return pRes->code; // free all allocated resource } + pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); + // use the local variable for (int32_t j = 0; j < numOfSub; ++j) { SSqlObj *pSub = pSql->pSubs[j]; @@ -1947,7 +1974,6 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) { tscProcessSql(pSub); } - pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); return TSDB_CODE_SUCCESS; _error: diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index d15a0d7fcc39703cff2c6e5d74f377c33e63b6b4..259bcd4cbd2e63f52c33f072908ef647731d750e 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -562,10 +562,8 @@ int32_t tscGetDataBlockFromList(void* pHashList, SArray* pDataBlockList, int64_t return TSDB_CODE_SUCCESS; } -static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock) { +static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bool includeSchema) { // TODO: optimize this function, handle the case while binary is not presented - int len = 0; - STableMeta* pTableMeta = pTableDataBlock->pTableMeta; STableComInfo tinfo = tscGetTableInfo(pTableMeta); SSchema* pSchema = tscGetTableSchema(pTableMeta); @@ -575,16 +573,37 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock) { pDataBlock += sizeof(SSubmitBlk); int32_t flen = 0; // original total length of row - for (int32_t i = 0; i < tinfo.numOfColumns; ++i) { - flen += TYPE_BYTES[pSchema[i].type]; + + // schema needs to be included into the submit data block + if (includeSchema) { + int32_t numOfCols = tscGetNumOfColumns(pTableDataBlock->pTableMeta); + for(int32_t j = 0; j < numOfCols; ++j) { + STColumn* pCol = (STColumn*) pDataBlock; + pCol->colId = pSchema[j].colId; + pCol->type = pSchema[j].type; + pCol->bytes = pSchema[j].bytes; + pCol->offset = 0; + + pDataBlock += sizeof(STColumn); + flen += TYPE_BYTES[pSchema[j].type]; + } + + int32_t schemaSize = sizeof(STColumn) * numOfCols; + pBlock->schemaLen = schemaSize; + } else { + for (int32_t j = 0; j < tinfo.numOfColumns; ++j) { + flen += TYPE_BYTES[pSchema[j].type]; + } + + pBlock->schemaLen = 0; } char* p = pTableDataBlock->pData + sizeof(SSubmitBlk); - pBlock->len = 0; + pBlock->dataLen = 0; int32_t numOfRows = htons(pBlock->numOfRows); for (int32_t i = 0; i < numOfRows; ++i) { - SDataRow trow = (SDataRow)pDataBlock; + SDataRow trow = (SDataRow) pDataBlock; dataRowSetLen(trow, TD_DATA_ROW_HEAD_SIZE + flen); dataRowSetVersion(trow, pTableMeta->sversion); @@ -595,20 +614,21 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock) { p += pSchema[j].bytes; } - // p += pTableDataBlock->rowSize; pDataBlock += dataRowLen(trow); - pBlock->len += dataRowLen(trow); + pBlock->dataLen += dataRowLen(trow); } - len = pBlock->len; - pBlock->len = htonl(pBlock->len); + int32_t len = pBlock->dataLen + pBlock->schemaLen; + pBlock->dataLen = htonl(pBlock->dataLen); + pBlock->schemaLen = htonl(pBlock->schemaLen); + return len; } int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) { SSqlCmd* pCmd = &pSql->cmd; - // the expanded size when a row data is converted to SDataRow format + // the maximum expanded size in byte when a row-wise data is converted to SDataRow format const int32_t MAX_EXPAND_SIZE = TD_DATA_ROW_HEAD_SIZE + TYPE_BYTES[TSDB_DATA_TYPE_BINARY]; void* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false); @@ -617,7 +637,6 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) { size_t total = taosArrayGetSize(pTableDataBlockList); for (int32_t i = 0; i < total; ++i) { STableDataBlocks* pOneTableBlock = taosArrayGetP(pTableDataBlockList, i); - STableDataBlocks* dataBuf = NULL; int32_t ret = @@ -666,16 +685,17 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) { pBlocks->uid = htobe64(pBlocks->uid); pBlocks->sversion = htonl(pBlocks->sversion); pBlocks->numOfRows = htons(pBlocks->numOfRows); + pBlocks->schemaLen = 0; // erase the empty space reserved for binary data - int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock); + int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pCmd->submitSchema); assert(finalLen <= len); dataBuf->size += (finalLen + sizeof(SSubmitBlk)); assert(dataBuf->size <= dataBuf->nAllocSize); // the length does not include the SSubmitBlk structure - pBlocks->len = htonl(finalLen); + pBlocks->dataLen = htonl(finalLen); dataBuf->numOfTables += 1; } diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h index 78255a45d5432fffe3ea0d2daf87d94dff041e42..d4a73e0e33292cff8cd19395bc25d799c7bb9194 100644 --- a/src/inc/taosmsg.h +++ b/src/inc/taosmsg.h @@ -192,7 +192,8 @@ typedef struct SSubmitBlk { int32_t tid; // table id int32_t padding; // TODO just for padding here int32_t sversion; // data schema version - int32_t len; // data part length, not including the SSubmitBlk head + int32_t dataLen; // data part length, not including the SSubmitBlk head + int32_t schemaLen; // schema length, if length is 0, no schema exists int16_t numOfRows; // total number of rows in current submit block char data[]; } SSubmitBlk; diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c index 6c6b3130737f2efeac3996b15aacb37133927176..e650cef45cf14a351ff7986985df43b4d1112b71 100644 --- a/src/tsdb/src/tsdbMain.c +++ b/src/tsdb/src/tsdbMain.c @@ -768,7 +768,8 @@ static SSubmitBlk *tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter) { SSubmitBlk *pBlock = pIter->pBlock; if (pBlock == NULL) return NULL; - pBlock->len = htonl(pBlock->len); + pBlock->dataLen = htonl(pBlock->dataLen); + pBlock->schemaLen = htonl(pBlock->schemaLen); pBlock->numOfRows = htons(pBlock->numOfRows); pBlock->uid = htobe64(pBlock->uid); pBlock->tid = htonl(pBlock->tid); @@ -776,11 +777,11 @@ static SSubmitBlk *tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter) { pBlock->sversion = htonl(pBlock->sversion); pBlock->padding = htonl(pBlock->padding); - pIter->len = pIter->len + sizeof(SSubmitBlk) + pBlock->len; + pIter->len = pIter->len + sizeof(SSubmitBlk) + pBlock->dataLen; if (pIter->len >= pIter->totalLen) { pIter->pBlock = NULL; } else { - pIter->pBlock = (SSubmitBlk *)((char *)pBlock + pBlock->len + sizeof(SSubmitBlk)); + pIter->pBlock = (SSubmitBlk *)((char *)pBlock + pBlock->dataLen + sizeof(SSubmitBlk)); } return pBlock; @@ -832,10 +833,10 @@ _err: } static int tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) { - if (pBlock->len <= 0) return -1; - pIter->totalLen = pBlock->len; + if (pBlock->dataLen <= 0) return -1; + pIter->totalLen = pBlock->dataLen; pIter->len = 0; - pIter->row = (SDataRow)(pBlock->data); + pIter->row = (SDataRow)(pBlock->data+pBlock->schemaLen); return 0; }