From 36d126c7cec90907b7d4dc0fe13e613aeab7c924 Mon Sep 17 00:00:00 2001 From: Hui Li Date: Tue, 22 Sep 2020 12:51:37 +0800 Subject: [PATCH 1/5] update version.inc to 2.0.4.0 --- cmake/version.inc | 2 +- src/connector/go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/cmake/version.inc b/cmake/version.inc index 52d62fca65..aa8a4b6463 100644 --- a/cmake/version.inc +++ b/cmake/version.inc @@ -4,7 +4,7 @@ PROJECT(TDengine) IF (DEFINED VERNUMBER) SET(TD_VER_NUMBER ${VERNUMBER}) ELSE () - SET(TD_VER_NUMBER "2.0.3.0") + SET(TD_VER_NUMBER "2.0.4.0") ENDIF () IF (DEFINED VERCOMPATIBLE) diff --git a/src/connector/go b/src/connector/go index 8c58c512b6..567b7b12f3 160000 --- a/src/connector/go +++ b/src/connector/go @@ -1 +1 @@ -Subproject commit 8c58c512b6acda8bcdfa48fdc7140227b5221766 +Subproject commit 567b7b12f3fd2775c718d284beffc8c38dd6c219 From a5f68b0ff8722575c1af901fd8e49319cd0a8b2d Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 25 Sep 2020 14:22:49 +0800 Subject: [PATCH 2/5] [td-1604]. --- src/client/src/tscParseInsert.c | 44 ++++++++++++++++----------------- src/client/src/tscPrepare.c | 6 ----- 2 files changed, 21 insertions(+), 29 deletions(-) diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index 327aac22d1..c7f168fa59 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -406,7 +406,7 @@ static int32_t tsCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start return TSDB_CODE_SUCCESS; } -int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[], SParsedDataColInfo *spd, char *error, +int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[], SParsedDataColInfo *spd, SSqlCmd* pCmd, int16_t timePrec, int32_t *code, char *tmpTokenBuf) { int32_t index = 0; SStrToken sToken = {0}; @@ -426,12 +426,17 @@ int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[ *str += index; if (sToken.type == TK_QUESTION) { + if (pCmd->insertType != TSDB_QUERY_TYPE_STMT_INSERT) { + *code = tscSQLSyntaxErrMsg(pCmd->payload, "? only allowed in binding insertion", *str); + return -1; + } + uint32_t offset = (uint32_t)(start - pDataBlocks->pData); if (tscAddParamToDataBlock(pDataBlocks, pSchema->type, (uint8_t)timePrec, pSchema->bytes, offset) != NULL) { continue; } - strcpy(error, "client out of memory"); + strcpy(pCmd->payload, "client out of memory"); *code = TSDB_CODE_TSC_OUT_OF_MEMORY; return -1; } @@ -439,8 +444,7 @@ int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[ int16_t type = sToken.type; if ((type != TK_NOW && type != TK_INTEGER && type != TK_STRING && type != TK_FLOAT && type != TK_BOOL && type != TK_NULL && type != TK_HEX && type != TK_OCT && type != TK_BIN) || (sToken.n == 0) || (type == TK_RP)) { - tscSQLSyntaxErrMsg(error, "invalid data or symbol", sToken.z); - *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR; + *code = tscSQLSyntaxErrMsg(pCmd->payload, "invalid data or symbol", sToken.z); return -1; } @@ -470,14 +474,14 @@ int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[ } bool isPrimaryKey = (colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX); - int32_t ret = tsParseOneColumnData(pSchema, &sToken, start, error, str, isPrimaryKey, timePrec); + int32_t ret = tsParseOneColumnData(pSchema, &sToken, start, pCmd->payload, str, isPrimaryKey, timePrec); if (ret != TSDB_CODE_SUCCESS) { *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR; return -1; // NOTE: here 0 mean error! } if (isPrimaryKey && tsCheckTimestamp(pDataBlocks, start) != TSDB_CODE_SUCCESS) { - tscInvalidSQLErrMsg(error, "client time/server time can not be mixed up", sToken.z); + tscInvalidSQLErrMsg(pCmd->payload, "client time/server time can not be mixed up", sToken.z); *code = TSDB_CODE_TSC_INVALID_TIME_STAMP; return -1; } @@ -522,7 +526,7 @@ static int32_t rowDataCompar(const void *lhs, const void *rhs) { } int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMeta, int maxRows, - SParsedDataColInfo *spd, char *error, int32_t *code, char *tmpTokenBuf) { + SParsedDataColInfo *spd, SSqlCmd* pCmd, int32_t *code, char *tmpTokenBuf) { int32_t index = 0; SStrToken sToken; @@ -534,8 +538,7 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMe int32_t precision = tinfo.precision; if (spd->hasVal[0] == false) { - strcpy(error, "primary timestamp column can not be null"); - *code = TSDB_CODE_TSC_INVALID_SQL; + *code = tscInvalidSQLErrMsg(pCmd->payload, "primary timestamp column can not be null", *str); return -1; } @@ -547,17 +550,17 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMe *str += index; if (numOfRows >= maxRows || pDataBlock->size + tinfo.rowSize >= pDataBlock->nAllocSize) { int32_t tSize; - int32_t retcode = tscAllocateMemIfNeed(pDataBlock, tinfo.rowSize, &tSize); - if (retcode != TSDB_CODE_SUCCESS) { //TODO pass the correct error code to client - strcpy(error, "client out of memory"); - *code = retcode; + *code = tscAllocateMemIfNeed(pDataBlock, tinfo.rowSize, &tSize); + if (*code != TSDB_CODE_SUCCESS) { //TODO pass the correct error code to client + strcpy(pCmd->payload, "client out of memory"); return -1; } + ASSERT(tSize > maxRows); maxRows = tSize; } - int32_t len = tsParseOneRowData(str, pDataBlock, pSchema, spd, error, precision, code, tmpTokenBuf); + int32_t len = tsParseOneRowData(str, pDataBlock, pSchema, spd, pCmd, precision, code, tmpTokenBuf); if (len <= 0) { // error message has been set in tsParseOneRowData return -1; } @@ -568,7 +571,7 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMe sToken = tStrGetToken(*str, &index, false, 0, NULL); *str += index; if (sToken.n == 0 || sToken.type != TK_RP) { - tscSQLSyntaxErrMsg(error, ") expected", *str); + tscSQLSyntaxErrMsg(pCmd->payload, ") expected", *str); *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR; return -1; } @@ -577,7 +580,7 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMe } if (numOfRows <= 0) { - strcpy(error, "no any data points"); + strcpy(pCmd->payload, "no any data points"); *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR; return -1; } else { @@ -704,7 +707,7 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableList, char **st return TSDB_CODE_TSC_OUT_OF_MEMORY; } - int32_t numOfRows = tsParseValues(str, dataBuf, pTableMeta, maxNumOfRows, spd, pCmd->payload, &code, tmpTokenBuf); + int32_t numOfRows = tsParseValues(str, dataBuf, pTableMeta, maxNumOfRows, spd, pCmd, &code, tmpTokenBuf); free(tmpTokenBuf); if (numOfRows <= 0) { return code; @@ -724,10 +727,6 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableList, char **st dataBuf->vgId = pTableMeta->vgroupInfo.vgId; dataBuf->numOfTables = 1; - /* - * the value of pRes->numOfRows does not affect the true result of AFFECTED ROWS, - * which is actually returned from server. - */ *totalNum += numOfRows; return TSDB_CODE_SUCCESS; } @@ -1460,8 +1459,7 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int code) { char *lineptr = line; strtolower(line, line); - int32_t len = - tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd->payload, tinfo.precision, &code, tokenBuf); + int32_t len = tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd, tinfo.precision, &code, tokenBuf); if (len <= 0 || pTableDataBlock->numOfParams > 0) { pSql->res.code = code; break; diff --git a/src/client/src/tscPrepare.c b/src/client/src/tscPrepare.c index c4ca6793ff..73425a81b3 100644 --- a/src/client/src/tscPrepare.c +++ b/src/client/src/tscPrepare.c @@ -43,10 +43,6 @@ typedef struct SNormalStmt { tVariant* params; } SNormalStmt; -//typedef struct SInsertStmt { -// -//} SInsertStmt; - typedef struct STscStmt { bool isInsert; STscObj* taos; @@ -54,7 +50,6 @@ typedef struct STscStmt { SNormalStmt normal; } STscStmt; - static int normalStmtAddPart(SNormalStmt* stmt, bool isParam, char* str, uint32_t len) { uint16_t size = stmt->numParts + 1; if (size > stmt->sizeParts) { @@ -514,7 +509,6 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) { SSqlObj* pSql = pStmt->pSql; size_t sqlLen = strlen(sql); - //doAsyncQuery(pObj, pSql, waitForQueryRsp, taos, sqlstr, sqlLen); SSqlCmd *pCmd = &pSql->cmd; SSqlRes *pRes = &pSql->res; pSql->param = (void*) pSql; From f82c9afcbcde398aaabaea6f7b05b24073f326a8 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sun, 27 Sep 2020 06:41:22 +0000 Subject: [PATCH 3/5] hotfix for invalid table id --- src/mnode/src/mnodeInt.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/mnode/src/mnodeInt.c b/src/mnode/src/mnodeInt.c index 91c8dcb6e5..fb1b8741a9 100644 --- a/src/mnode/src/mnodeInt.c +++ b/src/mnode/src/mnodeInt.c @@ -39,6 +39,11 @@ void mnodeCreateMsg(SMnodeMsg *pMsg, SRpcMsg *rpcMsg) { } int32_t mnodeInitMsg(SMnodeMsg *pMsg) { + if (pMsg->pUser != NULL) { + mDebug("app:%p:%p, user info already inited", pMsg->rpcMsg.ahandle, pMsg); + return TSDB_CODE_SUCCESS; + } + pMsg->pUser = mnodeGetUserFromConn(pMsg->rpcMsg.handle); if (pMsg->pUser == NULL) { return TSDB_CODE_MND_INVALID_USER; From 9c7aecc3d0ea8f4e9637b09835911712ba78299b Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Sun, 27 Sep 2020 15:00:35 +0800 Subject: [PATCH 4/5] [td-1619] --- src/tsdb/src/tsdbRead.c | 41 +++++++++++++++++++++++++++++++++-------- 1 file changed, 33 insertions(+), 8 deletions(-) diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index a3bc0de272..6c1602a857 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -697,22 +697,41 @@ static int32_t doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* p pCheckInfo->pDataCols = tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock); if (pCheckInfo->pDataCols == NULL) { - tsdbError("%p failed to malloc buf, %p", pQueryHandle, pQueryHandle->qinfo); + tsdbError("%p failed to malloc buf for pDataCols, %p", pQueryHandle, pQueryHandle->qinfo); terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; - return terrno; + goto _error; } } STSchema* pSchema = tsdbGetTableSchema(pCheckInfo->pTableObj); - tdInitDataCols(pCheckInfo->pDataCols, pSchema); - tdInitDataCols(pQueryHandle->rhelper.pDataCols[0], pSchema); - tdInitDataCols(pQueryHandle->rhelper.pDataCols[1], pSchema); + int32_t code = tdInitDataCols(pCheckInfo->pDataCols, pSchema); + if (code != TSDB_CODE_SUCCESS) { + tsdbError("%p failed to malloc buf for pDataCols, %p", pQueryHandle, pQueryHandle->qinfo); + terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; + goto _error; + } + + code = tdInitDataCols(pQueryHandle->rhelper.pDataCols[0], pSchema); + if (code != TSDB_CODE_SUCCESS) { + tsdbError("%p failed to malloc buf for rhelper.pDataCols[0], %p", pQueryHandle, pQueryHandle->qinfo); + terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; + goto _error; + } + + code = tdInitDataCols(pQueryHandle->rhelper.pDataCols[1], pSchema); + if (code != TSDB_CODE_SUCCESS) { + tsdbError("%p failed to malloc buf for rhelper.pDataCols[1], %p", pQueryHandle, pQueryHandle->qinfo); + terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; + goto _error; + } int16_t* colIds = pQueryHandle->defaultLoadColumn->pData; int32_t ret = tsdbLoadBlockDataCols(&(pQueryHandle->rhelper), pBlock, pCheckInfo->pCompInfo, colIds, (int)(QH_GET_NUM_OF_COLS(pQueryHandle))); if (ret != TSDB_CODE_SUCCESS) { - return terrno; + int32_t c = terrno; + assert(c != TSDB_CODE_SUCCESS); + goto _error; } SDataBlockLoadInfo* pBlockLoadInfo = &pQueryHandle->dataBlockLoadInfo; @@ -729,10 +748,16 @@ static int32_t doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* p int64_t elapsedTime = (taosGetTimestampUs() - st); pQueryHandle->cost.blockLoadTime += elapsedTime; - tsdbDebug("%p load file block into buffer, index:%d, brange:%"PRId64"-%"PRId64" , rows:%d, elapsed time:%"PRId64 " us, %p", + tsdbDebug("%p load file block into buffer, index:%d, brange:%"PRId64"-%"PRId64", rows:%d, elapsed time:%"PRId64 " us, %p", pQueryHandle, slotIndex, pBlock->keyFirst, pBlock->keyLast, pBlock->numOfRows, elapsedTime, pQueryHandle->qinfo); - return TSDB_CODE_SUCCESS; + +_error: + pBlock->numOfRows = 0; + + tsdbError("%p error occurs in loading file block, index:%d, brange:%"PRId64"-%"PRId64", rows:%d, %p", + pQueryHandle, slotIndex, pBlock->keyFirst, pBlock->keyLast, pBlock->numOfRows, pQueryHandle->qinfo); + return terrno; } static int32_t getEndPosInDataBlock(STsdbQueryHandle* pQueryHandle, SDataBlockInfo* pBlockInfo); From 2f346de95ebd7c4d0186cbfa8a2cf657ed516182 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Tue, 29 Sep 2020 03:16:47 +0000 Subject: [PATCH 5/5] Fix TD-1635 --- src/inc/taoserror.h | 1 + src/mnode/src/mnodeSdb.c | 6 ++++- src/wal/src/walMain.c | 48 ++++++++++++++++++++++++++++++++-------- 3 files changed, 45 insertions(+), 10 deletions(-) diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index 67e2d43c98..17b2d24e90 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -246,6 +246,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_ENABLED, 0, 0x0901, "Sync modul // wal TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR, 0, 0x1000, "Unexpected generic error in wal") +TAOS_DEFINE_ERROR(TSDB_CODE_WAL_FILE_CORRUPTED, 0, 0x1001, "WAL file is corrupted") // http TAOS_DEFINE_ERROR(TSDB_CODE_HTTP_SERVER_OFFLINE, 0, 0x1100, "http server is not onlin") diff --git a/src/mnode/src/mnodeSdb.c b/src/mnode/src/mnodeSdb.c index 646c17b2b8..4c672eb557 100644 --- a/src/mnode/src/mnodeSdb.c +++ b/src/mnode/src/mnodeSdb.c @@ -185,7 +185,11 @@ static int32_t sdbInitWal() { } sdbInfo("open sdb wal for restore"); - walRestore(tsSdbObj.wal, NULL, sdbWrite); + int code = walRestore(tsSdbObj.wal, NULL, sdbWrite); + if (code != TSDB_CODE_SUCCESS) { + sdbError("failed to open wal for restore, reason:%s", tstrerror(code)); + return -1; + } return 0; } diff --git a/src/wal/src/walMain.c b/src/wal/src/walMain.c index bebad69f32..4987ba2116 100644 --- a/src/wal/src/walMain.c +++ b/src/wal/src/walMain.c @@ -347,9 +347,10 @@ static void walRelease(SWal *pWal) { static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) { char *name = pWal->name; + int size = 1024 * 1024; // default 1M buffer size terrno = 0; - char *buffer = malloc(1024000); // size for one record + char *buffer = malloc(size); if (buffer == NULL) { terrno = TAOS_SYSTEM_ERROR(errno); return terrno; @@ -357,7 +358,7 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) { SWalHead *pHead = (SWalHead *)buffer; - int fd = open(name, O_RDONLY); + int fd = open(name, O_RDWR); if (fd < 0) { wError("wal:%s, failed to open for restore(%s)", name, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); @@ -367,29 +368,58 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) { wDebug("wal:%s, start to restore", name); + size_t offset = 0; while (1) { int ret = taosTRead(fd, pHead, sizeof(SWalHead)); - if ( ret == 0) break; + if (ret == 0) break; - if (ret != sizeof(SWalHead)) { - wWarn("wal:%s, failed to read head, skip, ret:%d(%s)", name, ret, strerror(errno)); + if (ret < 0) { + wError("wal:%s, failed to read wal head part since %s", name, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); break; } + if (ret < sizeof(SWalHead)) { + wError("wal:%s, failed to read head, ret:%d, skip the rest of file", name, ret); + taosFtruncate(fd, offset); + fsync(fd); + break; + } + if (!taosCheckChecksumWhole((uint8_t *)pHead, sizeof(SWalHead))) { wWarn("wal:%s, cksum is messed up, skip the rest of file", name); - terrno = TAOS_SYSTEM_ERROR(errno); + terrno = TSDB_CODE_WAL_FILE_CORRUPTED; + ASSERT(false); break; - } + } + + if (pHead->len > size - sizeof(SWalHead)) { + size = sizeof(SWalHead) + pHead->len; + buffer = realloc(buffer, size); + if (buffer == NULL) { + terrno = TAOS_SYSTEM_ERROR(errno); + break; + } + + pHead = (SWalHead *)buffer; + } ret = taosTRead(fd, pHead->cont, pHead->len); - if ( ret != pHead->len) { - wWarn("wal:%s, failed to read body, skip, len:%d ret:%d", name, pHead->len, ret); + if (ret < 0) { + wError("wal:%s failed to read wal body part since %s", name, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); break; } + if (ret < pHead->len) { + wError("wal:%s, failed to read body, len:%d ret:%d, skip the rest of file", name, pHead->len, ret); + taosFtruncate(fd, offset); + fsync(fd); + break; + } + + offset = offset + sizeof(SWalHead) + pHead->len; + if (pWal->keep) pWal->version = pHead->version; (*writeFp)(pVnode, pHead, TAOS_QTYPE_WAL); }