diff --git a/cmake/bdb_CMakeLists.txt.in b/cmake/bdb_CMakeLists.txt.in deleted file mode 100644 index dd21020105..0000000000 --- a/cmake/bdb_CMakeLists.txt.in +++ /dev/null @@ -1,13 +0,0 @@ - -# bdb -ExternalProject_Add(bdb - GIT_REPOSITORY https://github.com/berkeleydb/libdb.git - GIT_TAG v5.3.28 - SOURCE_DIR "${TD_CONTRIB_DIR}/bdb" - BINARY_DIR "${TD_CONTRIB_DIR}/bdb" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND COMMAND ./dist/configure --enable-debug - BUILD_COMMAND "$(MAKE)" - INSTALL_COMMAND "" - TEST_COMMAND "" -) diff --git a/cmake/leveldb_CMakeLists.txt.in b/cmake/leveldb_CMakeLists.txt.in deleted file mode 100644 index 6878711bc1..0000000000 --- a/cmake/leveldb_CMakeLists.txt.in +++ /dev/null @@ -1,13 +0,0 @@ - -# leveldb -ExternalProject_Add(leveldb - GIT_REPOSITORY https://github.com/taosdata-contrib/leveldb.git - GIT_TAG master - SOURCE_DIR "${TD_CONTRIB_DIR}/leveldb" - BINARY_DIR "" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" -) \ No newline at end of file diff --git a/cmake/lucene_CMakeLists.txt.in b/cmake/lucene_CMakeLists.txt.in deleted file mode 100644 index 9fd7471705..0000000000 --- a/cmake/lucene_CMakeLists.txt.in +++ /dev/null @@ -1,12 +0,0 @@ - -# lucene -ExternalProject_Add(lucene - GIT_REPOSITORY https://github.com/yihaoDeng/LucenePlusPlus.git - SOURCE_DIR "${TD_CONTRIB_DIR}/lucene" - BINARY_DIR "" - #BUILD_IN_SOURCE TRUE - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" -) diff --git a/cmake/nuraft_CMakeLists.txt.in b/cmake/nuraft_CMakeLists.txt.in deleted file mode 100644 index 593c6fed26..0000000000 --- a/cmake/nuraft_CMakeLists.txt.in +++ /dev/null @@ -1,12 +0,0 @@ - -# NuRaft -ExternalProject_Add(NuRaft - GIT_REPOSITORY https://github.com/eBay/NuRaft.git - GIT_TAG v1.3.0 - SOURCE_DIR "${TD_CONTRIB_DIR}/nuraft" - BINARY_DIR "${TD_CONTRIB_DIR}/nuraft" - CONFIGURE_COMMAND "./prepare.sh" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" - ) \ No newline at end of file diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt index 1b0a091e9d..bb18516ba4 100644 --- a/contrib/CMakeLists.txt +++ b/contrib/CMakeLists.txt @@ -109,11 +109,6 @@ cat("${TD_SUPPORT_DIR}/zlib_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) # cJson cat("${TD_SUPPORT_DIR}/cjson_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -# leveldb -if(${BUILD_WITH_LEVELDB}) - cat("${TD_SUPPORT_DIR}/leveldb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_LEVELDB}) - if (${BUILD_CONTRIB}) if(${BUILD_WITH_ROCKSDB}) cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -132,28 +127,11 @@ else() endif() endif() -# canonical-raft -if(${BUILD_WITH_CRAFT}) - cat("${TD_SUPPORT_DIR}/craft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - SET(BUILD_WITH_UV ON CACHE BOOL "craft need libuv" FORCE) -endif(${BUILD_WITH_CRAFT}) - -# traft -if(${BUILD_WITH_TRAFT}) - cat("${TD_SUPPORT_DIR}/traft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - SET(BUILD_WITH_UV ON CACHE BOOL "traft need libuv" FORCE) -endif(${BUILD_WITH_TRAFT}) - #libuv if(${BUILD_WITH_UV}) cat("${TD_SUPPORT_DIR}/libuv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) endif(${BUILD_WITH_UV}) -# bdb -if(${BUILD_WITH_BDB}) - cat("${TD_SUPPORT_DIR}/bdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_BDB}) - # sqlite if(${BUILD_WITH_SQLITE}) cat("${TD_SUPPORT_DIR}/sqlite_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -178,17 +156,6 @@ elseif(${BUILD_WITH_COS}) endif() -# lucene -if(${BUILD_WITH_LUCENE}) - cat("${TD_SUPPORT_DIR}/lucene_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) - add_definitions(-DUSE_LUCENE) -endif(${BUILD_WITH_LUCENE}) - -# NuRaft -if(${BUILD_WITH_NURAFT}) - cat("${TD_SUPPORT_DIR}/nuraft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) -endif(${BUILD_WITH_NURAFT}) - # crashdump if(${BUILD_CRASHDUMP}) cat("${TD_SUPPORT_DIR}/crashdump_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -448,23 +415,6 @@ elseif(${BUILD_WITH_COS}) endif() -# lucene -# To support build on ubuntu: sudo apt-get install libboost-all-dev -if(${BUILD_WITH_LUCENE}) - option(ENABLE_TEST "Enable the tests" OFF) - add_subdirectory(lucene EXCLUDE_FROM_ALL) - target_include_directories( - lucene++ - PUBLIC $ - ) - -endif(${BUILD_WITH_LUCENE}) - -# NuRaft -if(${BUILD_WITH_NURAFT}) - add_subdirectory(nuraft EXCLUDE_FROM_ALL) -endif(${BUILD_WITH_NURAFT}) - # pthread if(${BUILD_PTHREAD}) if ("${CMAKE_BUILD_TYPE}" STREQUAL "") @@ -537,30 +487,6 @@ if(${BUILD_WCWIDTH}) SET_TARGET_PROPERTIES(wcwidth PROPERTIES OUTPUT_NAME wcwidth) endif(${BUILD_WCWIDTH}) -# CRAFT -if(${BUILD_WITH_CRAFT}) - add_library(craft STATIC IMPORTED GLOBAL) - set_target_properties(craft PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/craft/.libs/libraft.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/craft/include" - ) - # target_link_libraries(craft - # INTERFACE pthread - # ) -endif(${BUILD_WITH_CRAFT}) - -# TRAFT -if(${BUILD_WITH_TRAFT}) - add_library(traft STATIC IMPORTED GLOBAL) - set_target_properties(traft PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/traft/.libs/libraft.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/traft/include" - ) - # target_link_libraries(craft - # INTERFACE pthread - # ) -endif(${BUILD_WITH_TRAFT}) - # LIBUV if(${BUILD_WITH_UV}) if (TD_WINDOWS) @@ -572,18 +498,6 @@ if(${BUILD_WITH_UV}) add_subdirectory(libuv EXCLUDE_FROM_ALL) endif(${BUILD_WITH_UV}) -# BDB -if(${BUILD_WITH_BDB}) - add_library(bdb STATIC IMPORTED GLOBAL) - set_target_properties(bdb PROPERTIES - IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/bdb/libdb.a" - INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_CURRENT_SOURCE_DIR}/bdb" - ) - target_link_libraries(bdb - INTERFACE pthread - ) -endif(${BUILD_WITH_BDB}) - # SQLite # see https://stackoverflow.com/questions/8774593/cmake-link-to-external-library#comment58570736_10550334 if(${BUILD_WITH_SQLITE}) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index e17a72992c..73d1ab2473 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -3271,7 +3271,7 @@ typedef struct { SMsgHead head; int64_t streamId; int32_t taskId; -} SVPauseStreamTaskReq, SVResetStreamTaskReq; +} SVPauseStreamTaskReq, SVResetStreamTaskReq, SVDropHTaskReq; typedef struct { int8_t reserved; diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index ba0217813c..95526222dc 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -271,6 +271,7 @@ enum { // WARN: new msg should be appended to segment tail TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_PAUSE, "stream-task-pause", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RESUME, "stream-task-resume", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_STOP, "stream-task-stop", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_STREAM_HTASK_DROP, "stream-htask-drop", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_STREAM_MAX_MSG, "stream-max", NULL, NULL) TD_NEW_MSG_SEG(TDMT_MON_MSG) //5 << 8 diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index f8ac360c71..e97f3e0192 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -659,17 +659,19 @@ int32_t tDecodeStreamCheckpointReadyMsg(SDecoder* pDecoder, SStreamCheckpointRea typedef struct STaskStatusEntry { STaskId id; int32_t status; + int32_t statusLastDuration; // to record the last duration of current status int64_t stage; int32_t nodeId; - int64_t verStart; // start version in WAL, only valid for source task - int64_t verEnd; // end version in WAL, only valid for source task - int64_t processedVer; // only valid for source task - int64_t activeCheckpointId; // current active checkpoint id - bool checkpointFailed; // denote if the checkpoint is failed or not - double inputQUsed; // in MiB + int64_t verStart; // start version in WAL, only valid for source task + int64_t verEnd; // end version in WAL, only valid for source task + int64_t processedVer; // only valid for source task + int32_t relatedHTask; // has related fill-history task + int64_t activeCheckpointId; // current active checkpoint id + bool checkpointFailed; // denote if the checkpoint is failed or not + double inputQUsed; // in MiB double inputRate; - double sinkQuota; // existed quota size for sink task - double sinkDataSize; // sink to dest data size + double sinkQuota; // existed quota size for sink task + double sinkDataSize; // sink to dst data size } STaskStatusEntry; typedef struct SStreamHbMsg { diff --git a/include/util/taoserror.h b/include/util/taoserror.h index ce8db162b6..0256a496df 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -124,6 +124,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_INVALID_CFG_VALUE TAOS_DEF_ERROR_CODE(0, 0x0133) #define TSDB_CODE_IP_NOT_IN_WHITE_LIST TAOS_DEF_ERROR_CODE(0, 0x0134) +#define TSDB_CODE_FAILED_TO_CONNECT_S3 TAOS_DEF_ERROR_CODE(0, 0x0135) //client #define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200) diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index da24bc0a3b..212c42125e 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -555,6 +555,9 @@ static void *tscCrashReportThreadFp(void *param) { if (pFile) { taosReleaseCrashLogFile(pFile, false); pFile = NULL; + + taosMsleep(sleepTime); + loopTimes = 0; continue; } } else { diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 6aebba7c02..d05cdc0156 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -1216,7 +1216,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { } int32_t retryCnt = 0; - while (TSDB_CODE_MND_CONSUMER_NOT_READY == syncAskEp(tmq)) { + while (syncAskEp(tmq) != 0) { if (retryCnt++ > MAX_RETRY_COUNT) { tscError("consumer:0x%" PRIx64 ", mnd not ready for subscribe, retry more than 2 minutes", tmq->consumerId); code = TSDB_CODE_MND_CONSUMER_NOT_READY; @@ -1454,6 +1454,9 @@ static bool doUpdateLocalEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp) if (epoch < tmq->epoch || (epoch == tmq->epoch && topicNumGet == 0)) { tscInfo("consumer:0x%" PRIx64 " no update ep epoch from %d to epoch %d, incoming topics:%d", tmq->consumerId, tmq->epoch, epoch, topicNumGet); + if(atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__RECOVER){ + atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY); + } return false; } @@ -1965,9 +1968,14 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) { return NULL; } - while (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__RECOVER) { + while (1) { + if(atomic_load_8(&tmq->status) != TMQ_CONSUMER_STATUS__RECOVER){ + break; + } + tscInfo("consumer:0x%" PRIx64 " tmq status is recover", tmq->consumerId); + int32_t retryCnt = 0; - while (TSDB_CODE_MND_CONSUMER_NOT_READY == syncAskEp(tmq)) { + while (syncAskEp(tmq) != 0) { if (retryCnt++ > 40) { return NULL; } diff --git a/source/common/src/cos.c b/source/common/src/cos.c index 1dca5eda9b..2a334a34b8 100644 --- a/source/common/src/cos.c +++ b/source/common/src/cos.c @@ -891,6 +891,8 @@ long s3Size(const char *object_name) { if ((cbd.status != S3StatusOK) && (cbd.status != S3StatusErrorPreconditionFailed)) { s3PrintError(__FILE__, __LINE__, __func__, cbd.status, cbd.err_msg); + + return -1; } size = cbd.content_length; diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index 18da1d638c..d6bdaf51bc 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -127,6 +127,9 @@ static void *dmCrashReportThreadFp(void *param) { if (pFile) { taosReleaseCrashLogFile(pFile, false); pFile = NULL; + + taosMsleep(sleepTime); + loopTimes = 0; continue; } } else { diff --git a/source/dnode/mgmt/mgmt_snode/src/smHandle.c b/source/dnode/mgmt/mgmt_snode/src/smHandle.c index cd81b9873f..6de29f8513 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smHandle.c +++ b/source/dnode/mgmt/mgmt_snode/src/smHandle.c @@ -84,6 +84,7 @@ SArray *smGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_HTASK_DROP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_SCAN_HISTORY_FINISH, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index c4c8a51d6b..a535ab17d7 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -833,6 +833,7 @@ SArray *vmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_HTASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_CHECK_POINT_SOURCE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECKPOINT_READY, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_UPDATE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index eb2f9dcbe3..1ddd2f34e6 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -1014,6 +1014,20 @@ static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *col return -1; } +static bool mndValidateSchema(SSchema *pSchemas, int32_t nSchema, SArray *pFields, int32_t maxLen) { + int32_t rowLen = 0; + for (int32_t i = 0; i < nSchema; ++i) { + rowLen += (pSchemas + i)->bytes; + } + + int32_t nField = taosArrayGetSize(pFields); + for (int32_t i = 0; i < nField; ++i) { + rowLen += ((SField *)TARRAY_GET_ELEM(pFields, i))->bytes; + } + + return rowLen <= maxLen; +} + static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) { taosRLockLatch(&pStb->lock); memcpy(pDst, pStb, sizeof(SStbObj)); @@ -1269,6 +1283,11 @@ static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *p return -1; } + if (!mndValidateSchema(pOld->pTags, pOld->numOfTags, pFields, TSDB_MAX_TAGS_LEN)) { + terrno = TSDB_CODE_PAR_INVALID_TAGS_LENGTH; + return -1; + } + pNew->numOfTags = pNew->numOfTags + ntags; if (mndAllocStbSchemas(pOld, pNew) != 0) { return -1; @@ -1558,6 +1577,16 @@ static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj return -1; } + uint32_t nLen = 0; + for (int32_t i = 0; i < pOld->numOfTags; ++i) { + nLen += (pOld->pTags[i].colId == colId) ? pField->bytes : pOld->pTags[i].bytes; + } + + if (nLen > TSDB_MAX_TAGS_LEN) { + terrno = TSDB_CODE_PAR_INVALID_TAGS_LENGTH; + return -1; + } + if (mndAllocStbSchemas(pOld, pNew) != 0) { return -1; } @@ -1592,6 +1621,11 @@ static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, SArray return -1; } + if (!mndValidateSchema(pOld->pColumns, pOld->numOfColumns, pFields, TSDB_MAX_BYTES_PER_ROW)) { + terrno = TSDB_CODE_PAR_INVALID_ROW_LENGTH; + return -1; + } + pNew->numOfColumns = pNew->numOfColumns + ncols; if (mndAllocStbSchemas(pOld, pNew) != 0) { return -1; diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 0167c1a6ac..235703428f 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -79,6 +79,8 @@ static int32_t mndProcessNodeCheckReq(SRpcMsg *pMsg); static SArray *extractNodeListFromStream(SMnode *pMnode); static SArray *mndTakeVgroupSnapshot(SMnode *pMnode, bool *allReady); +static SStreamObj *mndGetStreamObj(SMnode *pMnode, int64_t streamId); + static SVgroupChangeInfo mndFindChangedNodeInfo(SMnode *pMnode, const SArray *pPrevNodeList, const SArray *pNodeList); static STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, SRpcMsg *pReq, const char *name, const char* pMsg); @@ -698,17 +700,8 @@ static int32_t mndPersistTaskDropReq(SMnode* pMnode, STrans *pTrans, SStreamTask STransAction action = {0}; SEpSet epset = {0}; if(pTask->info.nodeId == SNODE_HANDLE){ - SSnodeObj *pObj = NULL; - void *pIter = NULL; - while (1) { - pIter = sdbFetch(pMnode->pSdb, SDB_SNODE, pIter, (void **)&pObj); - if (pIter == NULL) { - break; - } - - addEpIntoEpSet(&epset, pObj->pDnode->fqdn, pObj->pDnode->port); - sdbRelease(pMnode->pSdb, pObj); - } + SSnodeObj* pObj = mndAcquireSnode(pMnode, pTask->info.nodeId); + addEpIntoEpSet(&epset, pObj->pDnode->fqdn, pObj->pDnode->port); }else{ SVgObj *pVgObj = mndAcquireVgroup(pMnode, pTask->info.nodeId); epset = mndGetVgroupEpset(pMnode, pVgObj); @@ -1586,6 +1579,123 @@ static void mndCancelGetNextStream(SMnode *pMnode, void *pIter) { sdbCancelFetch(pSdb, pIter); } +static void setTaskAttrInResBlock(SStreamObj* pStream, SStreamTask* pTask, SSDataBlock* pBlock, int32_t numOfRows) { + SColumnInfoData *pColInfo; + int32_t cols = 0; + + // stream name + char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; + STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName)); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false); + + // task id + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + + char idstr[128] = {0}; + int32_t len = tintToHex(pTask->id.taskId, &idstr[4]); + idstr[2] = '0'; + idstr[3] = 'x'; + varDataSetLen(idstr, len + 2); + colDataSetVal(pColInfo, numOfRows, idstr, false); + + // node type + char nodeType[20 + VARSTR_HEADER_SIZE] = {0}; + varDataSetLen(nodeType, 5); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + if (pTask->info.nodeId > 0) { + memcpy(varDataVal(nodeType), "vnode", 5); + } else { + memcpy(varDataVal(nodeType), "snode", 5); + } + colDataSetVal(pColInfo, numOfRows, nodeType, false); + + // node id + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + int64_t nodeId = TMAX(pTask->info.nodeId, 0); + colDataSetVal(pColInfo, numOfRows, (const char *)&nodeId, false); + + // level + char level[20 + VARSTR_HEADER_SIZE] = {0}; + if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { + memcpy(varDataVal(level), "source", 6); + varDataSetLen(level, 6); + } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) { + memcpy(varDataVal(level), "agg", 3); + varDataSetLen(level, 3); + } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) { + memcpy(varDataVal(level), "sink", 4); + varDataSetLen(level, 4); + } + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)level, false); + + // status + char status[20 + VARSTR_HEADER_SIZE] = {0}; + STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId}; + + STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id)); + if (pe == NULL) { + return; + } + + const char *pStatus = streamTaskGetStatusStr(pe->status); + STR_TO_VARSTR(status, pStatus); + + // status + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)status, false); + + // stage + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)&pe->stage, false); + + // input queue + char vbuf[30] = {0}; + char buf[25] = {0}; + const char *queueInfoStr = "%4.2fMiB (%5.2f%)"; + sprintf(buf, queueInfoStr, pe->inputQUsed, pe->inputRate); + STR_TO_VARSTR(vbuf, buf); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); + + // output queue + // sprintf(buf, queueInfoStr, pe->outputQUsed, pe->outputRate); + // STR_TO_VARSTR(vbuf, buf); + + // pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + // colDataSetVal(pColInfo, numOfRows, (const char*)vbuf, false); + + if (pTask->info.taskLevel == TASK_LEVEL__SINK) { + const char *sinkStr = "%.2fMiB"; + sprintf(buf, sinkStr, pe->sinkDataSize); + } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { + // offset info + const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; + sprintf(buf, offsetStr, pe->processedVer, pe->verStart, pe->verEnd); + } + + STR_TO_VARSTR(vbuf, buf); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); +} + +static int32_t getNumOfTasks(SArray* pTaskList) { + int32_t numOfLevels = taosArrayGetSize(pTaskList); + + int32_t count = 0; + for (int32_t i = 0; i < numOfLevels; i++) { + SArray *pLevel = taosArrayGetP(pTaskList, i); + count += taosArrayGetSize(pLevel); + } + + return count; +} + static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) { SMnode * pMnode = pReq->info.node; SSdb * pSdb = pMnode->pSdb; @@ -1601,137 +1711,25 @@ static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock // lock taosRLockLatch(&pStream->lock); - // count task num - int32_t sz = taosArrayGetSize(pStream->tasks); - - int32_t count = 0; - for (int32_t i = 0; i < sz; i++) { - SArray *pLevel = taosArrayGetP(pStream->tasks, i); - count += taosArrayGetSize(pLevel); - } - + int32_t count = getNumOfTasks(pStream->tasks); if (numOfRows + count > rowsCapacity) { blockDataEnsureCapacity(pBlock, numOfRows + count); } // add row for each task - for (int32_t i = 0; i < sz; i++) { + for (int32_t i = 0; i < taosArrayGetSize(pStream->tasks); i++) { SArray *pLevel = taosArrayGetP(pStream->tasks, i); - int32_t levelCnt = taosArrayGetSize(pLevel); - for (int32_t j = 0; j < levelCnt; j++) { + int32_t numOfLevels = taosArrayGetSize(pLevel); + for (int32_t j = 0; j < numOfLevels; j++) { SStreamTask *pTask = taosArrayGetP(pLevel, j); - - SColumnInfoData *pColInfo; - int32_t cols = 0; - - // stream name - char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; - STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName)); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false); - - // task id - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - - char idstr[128] = {0}; - int32_t len = tintToHex(pTask->id.taskId, &idstr[4]); - idstr[2] = '0'; - idstr[3] = 'x'; - varDataSetLen(idstr, len + 2); - colDataSetVal(pColInfo, numOfRows, idstr, false); - - // node type - char nodeType[20 + VARSTR_HEADER_SIZE] = {0}; - varDataSetLen(nodeType, 5); - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - if (pTask->info.nodeId > 0) { - memcpy(varDataVal(nodeType), "vnode", 5); - } else { - memcpy(varDataVal(nodeType), "snode", 5); - } - colDataSetVal(pColInfo, numOfRows, nodeType, false); - - // node id - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - int64_t nodeId = TMAX(pTask->info.nodeId, 0); - colDataSetVal(pColInfo, numOfRows, (const char *)&nodeId, false); - - // level - char level[20 + VARSTR_HEADER_SIZE] = {0}; - if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - memcpy(varDataVal(level), "source", 6); - varDataSetLen(level, 6); - } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) { - memcpy(varDataVal(level), "agg", 3); - varDataSetLen(level, 3); - } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) { - memcpy(varDataVal(level), "sink", 4); - varDataSetLen(level, 4); - } - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)level, false); - - // status - char status[20 + VARSTR_HEADER_SIZE] = {0}; - - STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId}; - STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id)); - if (pe == NULL) { - continue; - } - - const char *pStatus = streamTaskGetStatusStr(pe->status); - STR_TO_VARSTR(status, pStatus); - - // status - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)status, false); - - // stage - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)&pe->stage, false); - - // input queue - char vbuf[30] = {0}; - char buf[25] = {0}; - const char *queueInfoStr = "%4.2fMiB (%5.2f%)"; - sprintf(buf, queueInfoStr, pe->inputQUsed, pe->inputRate); - STR_TO_VARSTR(vbuf, buf); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); - - // output queue - // sprintf(buf, queueInfoStr, pe->outputQUsed, pe->outputRate); - // STR_TO_VARSTR(vbuf, buf); - - // pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - // colDataSetVal(pColInfo, numOfRows, (const char*)vbuf, false); - - if (pTask->info.taskLevel == TASK_LEVEL__SINK) { - const char *sinkStr = "%.2fMiB"; - sprintf(buf, sinkStr, pe->sinkDataSize); - } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - // offset info - const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; - sprintf(buf, offsetStr, pe->processedVer, pe->verStart, pe->verEnd); - } - - STR_TO_VARSTR(vbuf, buf); - - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false); - + setTaskAttrInResBlock(pStream, pTask, pBlock, numOfRows); numOfRows++; } } // unlock taosRUnLockLatch(&pStream->lock); - sdbRelease(pSdb, pStream); } @@ -2745,7 +2743,7 @@ int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDBName, size_t le return TSDB_CODE_SUCCESS; } -int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { +static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { STrans* pTrans = mndAcquireTrans(pMnode, transId); if (pTrans != NULL) { mInfo("kill checkpoint transId:%d to reset task status", transId); @@ -2781,6 +2779,91 @@ int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) { return 0; } +static SStreamTask* mndGetStreamTask(STaskId* pId, SStreamObj* pStream) { + for (int32_t i = 0; i < taosArrayGetSize(pStream->tasks); i++) { + SArray *pLevel = taosArrayGetP(pStream->tasks, i); + + int32_t numOfLevels = taosArrayGetSize(pLevel); + for (int32_t j = 0; j < numOfLevels; j++) { + SStreamTask *pTask = taosArrayGetP(pLevel, j); + if (pTask->id.taskId == pId->taskId) { + return pTask; + } + } + } + + return NULL; +} + +static bool needDropRelatedFillhistoryTask(STaskStatusEntry *pTaskEntry, SStreamExecInfo *pExecNode) { + if (pTaskEntry->status == TASK_STATUS__STREAM_SCAN_HISTORY && pTaskEntry->statusLastDuration >= 10) { + if (fabs(pTaskEntry->inputQUsed) <= DBL_EPSILON) { + int32_t numOfReady = 0; + int32_t numOfTotal = 0; + for (int32_t k = 0; k < taosArrayGetSize(pExecNode->pTaskList); ++k) { + STaskId *pId = taosArrayGet(pExecNode->pTaskList, k); + if (pTaskEntry->id.streamId == pId->streamId) { + numOfTotal++; + + if (pTaskEntry->id.taskId != pId->taskId) { + STaskStatusEntry *pEntry = taosHashGet(execInfo.pTaskMap, pId, sizeof(*pId)); + if (pEntry->status == TASK_STATUS__READY) { + numOfReady++; + } + } + } + } + + if (numOfReady > 0) { + mDebug("stream:0x%" PRIx64 + " %d tasks are ready, %d tasks in stream-scan-history for more than 50s, drop related fill-history task", + pTaskEntry->id.streamId, numOfReady, numOfTotal - numOfReady); + return true; + } else { + return false; + } + } + } + + return false; +} + +// currently only handle the sink task +// 1. sink task, drop related fill-history task msg is missing +// 2. other tasks are in ready state for at least 3 * hb_interval +static int32_t mndDropRelatedFillhistoryTask(SMnode *pMnode, STaskStatusEntry *pTaskEntry, SStreamObj *pStream) { + SStreamTask *pTask = mndGetStreamTask(&pTaskEntry->id, pStream); + if (pTask == NULL) { + mError("failed to get the stream task:0x%x, may have been dropped", (int32_t) pTaskEntry->id.taskId); + return -1; + } + + SVDropHTaskReq *pReq = rpcMallocCont(sizeof(SVDropHTaskReq)); + if (pReq == NULL) { + mError("failed to malloc in drop related fill-history task, size:%" PRIzu ", code:%s", sizeof(SVDropHTaskReq), + tstrerror(TSDB_CODE_OUT_OF_MEMORY)); + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + pReq->head.vgId = htonl(pTask->info.nodeId); + pReq->taskId = pTask->id.taskId; + pReq->streamId = pTask->id.streamId; + + SRpcMsg msg = {.info.noResp = 1}; + + initRpcMsg(&msg, TDMT_STREAM_HTASK_DROP, pReq, sizeof(SVDropHTaskReq)); + + mDebug("build and send drop related fill-history task for task:0x%x", pTask->id.taskId); + + SVgObj *pVgObj = mndAcquireVgroup(pMnode, pTask->info.nodeId); + SEpSet epset = mndGetVgroupEpset(pMnode, pVgObj); + mndReleaseVgroup(pMnode, pVgObj); + + tmsgSendReq(&epset, &msg); + return TSDB_CODE_SUCCESS; +} + int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList) { int32_t num = taosArrayGetSize(pNodeList); mInfo("set node expired for %d nodes", num); @@ -2881,9 +2964,28 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) { } } - pTaskEntry->status = p->status; + if (p->status == pTaskEntry->status) { + pTaskEntry->statusLastDuration++; + } else { + pTaskEntry->status = p->status; + pTaskEntry->statusLastDuration = 0; + } + if (p->status != TASK_STATUS__READY) { mDebug("received s-task:0x%" PRIx64 " not in ready status:%s", p->id.taskId, streamTaskGetStatusStr(p->status)); + + if (p->status == TASK_STATUS__STREAM_SCAN_HISTORY) { + bool drop = needDropRelatedFillhistoryTask(pTaskEntry, &execInfo); + if(drop) { + SStreamObj *pStreamObj = mndGetStreamObj(pMnode, pTaskEntry->id.streamId); + if (pStreamObj == NULL) { + mError("failed to acquire the streamObj:0x%" PRIx64 " it may have been dropped", pStreamObj->uid); + } else { + mndDropRelatedFillhistoryTask(pMnode, pTaskEntry, pStreamObj); + mndReleaseStream(pMnode, pStreamObj); + } + } + } } } @@ -2910,3 +3012,18 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) { taosArrayDestroy(req.pUpdateNodes); return TSDB_CODE_SUCCESS; } + +SStreamObj *mndGetStreamObj(SMnode *pMnode, int64_t streamId) { + void *pIter = NULL; + SSdb *pSdb = pMnode->pSdb; + SStreamObj *pStream = NULL; + + while ((pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream)) != NULL) { + if (pStream->uid == streamId) { + sdbCancelFetch(pSdb, pIter); + return pStream; + } + } + + return NULL; +} \ No newline at end of file diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h index c74ccf6c11..7dbaa66d44 100644 --- a/source/dnode/vnode/src/inc/meta.h +++ b/source/dnode/vnode/src/inc/meta.h @@ -79,6 +79,7 @@ struct SMeta { char* path; SVnode* pVnode; + bool changed; TDB* pEnv; TXN* txn; TTB* pTbDb; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index a14e91d7e6..473329bfee 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -232,6 +232,7 @@ int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp) int32_t tqProcessTaskCheckpointReadyMsg(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg); +int32_t tqProcessTaskDropHTask(STQ* pTq, SRpcMsg* pMsg); int32_t tqRestartStreamTasks(STQ* pTq); int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver); diff --git a/source/dnode/vnode/src/meta/metaCommit.c b/source/dnode/vnode/src/meta/metaCommit.c index f5572e68dd..f8b41e413b 100644 --- a/source/dnode/vnode/src/meta/metaCommit.c +++ b/source/dnode/vnode/src/meta/metaCommit.c @@ -56,7 +56,7 @@ int metaPrepareAsyncCommit(SMeta *pMeta) { code = ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn); metaULock(pMeta); code = tdbCommit(pMeta->pEnv, pMeta->txn); - + pMeta->changed = false; return code; } diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index f600925a0b..976ee616f9 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -251,6 +251,7 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { ++pMeta->pVnode->config.vndStats.numOfSTables; + pMeta->changed = true; metaDebug("vgId:%d, stb:%s is created, suid:%" PRId64, TD_VID(pMeta->pVnode), pReq->name, pReq->suid); return 0; @@ -325,6 +326,8 @@ _drop_super_table: metaUpdTimeSeriesNum(pMeta); + pMeta->changed = true; + _exit: tdbFree(pKey); tdbFree(pData); @@ -424,6 +427,8 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { metaTimeSeriesNotifyCheck(pMeta); } + pMeta->changed = true; + _exit: if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf); tDecoderClear(&dc); @@ -847,6 +852,7 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs } } + pMeta->changed = true; metaDebug("vgId:%d, table:%s uid %" PRId64 " is created, type:%" PRId8, TD_VID(pMeta->pVnode), pReq->name, pReq->uid, pReq->type); return 0; @@ -895,6 +901,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi *tbUid = uid; } + pMeta->changed = true; _exit: tdbFree(pData); return rc; @@ -938,6 +945,8 @@ void metaDropTables(SMeta *pMeta, SArray *tbUids) { } } tSimpleHashCleanup(suidHash); + + pMeta->changed = true; } static int32_t metaFilterTableByHash(SMeta *pMeta, SArray *uidList) { @@ -1233,6 +1242,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl if (pAlterTbReq->colName == NULL) { terrno = TSDB_CODE_INVALID_MSG; + metaError("meta/table: null pAlterTbReq->colName"); return -1; } @@ -1300,20 +1310,27 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl SMetaEntry oldEntry = {.type = TSDB_NORMAL_TABLE, .uid = entry.uid}; oldEntry.ntbEntry.schemaRow.nCols = pSchema->nCols; - int32_t iCol = 0; + int32_t rowLen = -1; + if (pAlterTbReq->action == TSDB_ALTER_TABLE_ADD_COLUMN || + pAlterTbReq->action == TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES) { + rowLen = 0; + } + + int32_t iCol = 0, jCol = 0; + SSchema *qColumn = NULL; for (;;) { - pColumn = NULL; + qColumn = NULL; - if (iCol >= pSchema->nCols) break; - pColumn = &pSchema->pSchema[iCol]; + if (jCol >= pSchema->nCols) break; + qColumn = &pSchema->pSchema[jCol]; - if (NULL == pAlterTbReq->colName) { - metaError("meta/table: null pAlterTbReq->colName"); - return -1; + if (!pColumn && (strcmp(qColumn->name, pAlterTbReq->colName) == 0)) { + pColumn = qColumn; + iCol = jCol; + if (rowLen < 0) break; } - - if (strcmp(pColumn->name, pAlterTbReq->colName) == 0) break; - iCol++; + rowLen += qColumn->bytes; + ++jCol; } entry.version = version; @@ -1328,6 +1345,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) { goto _err; } + if (rowLen + pAlterTbReq->bytes > TSDB_MAX_BYTES_PER_ROW) { + terrno = TSDB_CODE_PAR_INVALID_ROW_LENGTH; + goto _err; + } pSchema->version++; pSchema->nCols++; pNewSchema = taosMemoryMalloc(sizeof(SSchema) * pSchema->nCols); @@ -1369,10 +1390,14 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl terrno = TSDB_CODE_VND_COL_NOT_EXISTS; goto _err; } - if (!IS_VAR_DATA_TYPE(pColumn->type) || pColumn->bytes > pAlterTbReq->colModBytes) { + if (!IS_VAR_DATA_TYPE(pColumn->type) || pColumn->bytes >= pAlterTbReq->colModBytes) { terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; goto _err; } + if (rowLen + pAlterTbReq->colModBytes - pColumn->bytes > TSDB_MAX_BYTES_PER_ROW) { + terrno = TSDB_CODE_PAR_INVALID_ROW_LENGTH; + goto _err; + } if (tqCheckColModifiable(pMeta->pVnode->pTq, uid, pColumn->colId) != 0) { terrno = TSDB_CODE_VND_COL_SUBSCRIBED; goto _err; @@ -1970,6 +1995,7 @@ _err: } int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq, STableMetaRsp *pMetaRsp) { + pMeta->changed = true; switch (pReq->action) { case TSDB_ALTER_TABLE_ADD_COLUMN: case TSDB_ALTER_TABLE_DROP_COLUMN: diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 5b04b31c94..e66c0357a1 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1388,3 +1388,34 @@ int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg) { streamMetaReleaseTask(pMeta, pTask); return TSDB_CODE_SUCCESS; } + +int32_t tqProcessTaskDropHTask(STQ* pTq, SRpcMsg* pMsg) { + SVDropHTaskReq* pReq = (SVDropHTaskReq*) pMsg->pCont; + + SStreamMeta* pMeta = pTq->pStreamMeta; + SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId); + if (pTask == NULL) { + tqError("vgId:%d process drop fill-history task req, failed to acquire task:0x%x, it may have been dropped already", + pMeta->vgId, pReq->taskId); + return TSDB_CODE_SUCCESS; + } + + tqDebug("s-task:%s receive drop fill-history msg from mnode", pTask->id.idStr); + if (pTask->hTaskInfo.id.taskId == 0) { + tqError("vgId:%d s-task:%s not have related fill-history task", pMeta->vgId, pTask->id.idStr); + streamMetaReleaseTask(pMeta, pTask); + return TSDB_CODE_SUCCESS; + } + + ETaskStatus status = streamTaskGetStatus(pTask, NULL); + ASSERT(status == TASK_STATUS__STREAM_SCAN_HISTORY); + + streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_SCANHIST_DONE); + + SStreamTaskId id = {.streamId = pTask->hTaskInfo.id.streamId, .taskId = pTask->hTaskInfo.id.taskId}; + streamBuildAndSendDropTaskMsg(pTask->pMsgCb, pMeta->vgId, &id); + + streamMetaReleaseTask(pMeta, pTask); + return TSDB_CODE_SUCCESS; +} + diff --git a/source/dnode/vnode/src/tsdb/tsdbCommit2.c b/source/dnode/vnode/src/tsdb/tsdbCommit2.c index 48b622e324..a974eb27bf 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCommit2.c +++ b/source/dnode/vnode/src/tsdb/tsdbCommit2.c @@ -421,7 +421,7 @@ static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) { if (mtime < committer->ctx->now - tsS3UploadDelaySec) { committer->ctx->skipTsRow = true; } - } else if (s3Size(object_name) > 0) { + } else /*if (s3Size(object_name) > 0) */ { committer->ctx->skipTsRow = true; } } diff --git a/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c b/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c index df6b85a889..e1625c9ddb 100644 --- a/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c +++ b/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c @@ -45,7 +45,7 @@ static int32_t tsdbDataFileReadHeadFooter(SDataFileReader *reader) { int32_t ftype = TSDB_FTYPE_HEAD; if (reader->fd[ftype]) { code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(SHeadFooter), - (uint8_t *)reader->headFooter, sizeof(SHeadFooter)); + (uint8_t *)reader->headFooter, sizeof(SHeadFooter), 0); TSDB_CHECK_CODE(code, lino, _exit); } @@ -67,7 +67,7 @@ static int32_t tsdbDataFileReadTombFooter(SDataFileReader *reader) { int32_t ftype = TSDB_FTYPE_TOMB; if (reader->fd[ftype]) { code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(STombFooter), - (uint8_t *)reader->tombFooter, sizeof(STombFooter)); + (uint8_t *)reader->tombFooter, sizeof(STombFooter), 0); TSDB_CHECK_CODE(code, lino, _exit); } reader->ctx->tombFooterLoaded = true; @@ -161,7 +161,7 @@ int32_t tsdbDataFileReadBrinBlk(SDataFileReader *reader, const TBrinBlkArray **b } code = tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], reader->headFooter->brinBlkPtr->offset, data, - reader->headFooter->brinBlkPtr->size); + reader->headFooter->brinBlkPtr->size, 0); if (code) { taosMemoryFree(data); TSDB_CHECK_CODE(code, lino, _exit); @@ -191,7 +191,8 @@ int32_t tsdbDataFileReadBrinBlock(SDataFileReader *reader, const SBrinBlk *brinB code = tRealloc(&reader->config->bufArr[0], brinBlk->dp->size); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], brinBlk->dp->offset, reader->config->bufArr[0], brinBlk->dp->size); + code = + tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], brinBlk->dp->offset, reader->config->bufArr[0], brinBlk->dp->size, 0); TSDB_CHECK_CODE(code, lino, _exit); int32_t size = 0; @@ -232,7 +233,8 @@ int32_t tsdbDataFileReadBlockData(SDataFileReader *reader, const SBrinRecord *re code = tRealloc(&reader->config->bufArr[0], record->blockSize); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockSize); + code = + tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockSize, 0); TSDB_CHECK_CODE(code, lino, _exit); code = tDecmprBlockData(reader->config->bufArr[0], record->blockSize, bData, &reader->config->bufArr[1]); @@ -257,8 +259,8 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe code = tRealloc(&reader->config->bufArr[0], record->blockKeySize); TSDB_CHECK_CODE(code, lino, _exit); - code = - tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockKeySize); + code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockKeySize, + 0); TSDB_CHECK_CODE(code, lino, _exit); // hdr @@ -296,10 +298,46 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe TSDB_CHECK_CODE(code, lino, _exit); code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset + record->blockKeySize, - reader->config->bufArr[0], hdr->szBlkCol); + reader->config->bufArr[0], hdr->szBlkCol, 0); TSDB_CHECK_CODE(code, lino, _exit); } + int64_t szHint = 0; + if (bData->nColData > 3) { + int64_t offset = 0; + SBlockCol bc = {.cid = 0}; + SBlockCol *blockCol = &bc; + + size = 0; + SColData *colData = tBlockDataGetColDataByIdx(bData, 0); + while (blockCol && blockCol->cid < colData->cid) { + if (size < hdr->szBlkCol) { + size += tGetBlockCol(reader->config->bufArr[0] + size, blockCol); + } else { + ASSERT(size == hdr->szBlkCol); + blockCol = NULL; + } + } + + if (blockCol && blockCol->flag == HAS_VALUE) { + offset = blockCol->offset; + + SColData *colDataEnd = tBlockDataGetColDataByIdx(bData, bData->nColData - 1); + while (blockCol && blockCol->cid < colDataEnd->cid) { + if (size < hdr->szBlkCol) { + size += tGetBlockCol(reader->config->bufArr[0] + size, blockCol); + } else { + ASSERT(size == hdr->szBlkCol); + blockCol = NULL; + } + } + + if (blockCol && blockCol->flag == HAS_VALUE) { + szHint = blockCol->offset + blockCol->szBitmap + blockCol->szOffset + blockCol->szValue - offset; + } + } + } + SBlockCol bc[1] = {{.cid = 0}}; SBlockCol *blockCol = bc; @@ -338,7 +376,7 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset + record->blockKeySize + hdr->szBlkCol + blockCol->offset, - reader->config->bufArr[1], size1); + reader->config->bufArr[1], size1, i > 0 ? 0 : szHint); TSDB_CHECK_CODE(code, lino, _exit); code = tsdbDecmprColData(reader->config->bufArr[1], blockCol, hdr->cmprAlg, hdr->nRow, colData, @@ -366,7 +404,7 @@ int32_t tsdbDataFileReadBlockSma(SDataFileReader *reader, const SBrinRecord *rec code = tRealloc(&reader->config->bufArr[0], record->smaSize); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd[TSDB_FTYPE_SMA], record->smaOffset, reader->config->bufArr[0], record->smaSize); + code = tsdbReadFile(reader->fd[TSDB_FTYPE_SMA], record->smaOffset, reader->config->bufArr[0], record->smaSize, 0); TSDB_CHECK_CODE(code, lino, _exit); // decode sma data @@ -405,7 +443,7 @@ int32_t tsdbDataFileReadTombBlk(SDataFileReader *reader, const TTombBlkArray **t } code = tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], reader->tombFooter->tombBlkPtr->offset, data, - reader->tombFooter->tombBlkPtr->size); + reader->tombFooter->tombBlkPtr->size, 0); if (code) { taosMemoryFree(data); TSDB_CHECK_CODE(code, lino, _exit); @@ -435,7 +473,8 @@ int32_t tsdbDataFileReadTombBlock(SDataFileReader *reader, const STombBlk *tombB code = tRealloc(&reader->config->bufArr[0], tombBlk->dp->size); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size); + code = + tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size, 0); TSDB_CHECK_CODE(code, lino, _exit); int32_t size = 0; @@ -488,8 +527,8 @@ struct SDataFileWriter { STombBlock tombBlock[1]; int32_t tombBlockIdx; // range - SVersionRange range; - SVersionRange tombRange; + SVersionRange range; + SVersionRange tombRange; } ctx[1]; STFile files[TSDB_FTYPE_MAX]; diff --git a/source/dnode/vnode/src/tsdb/tsdbDef.h b/source/dnode/vnode/src/tsdb/tsdbDef.h index da2445dee5..0f512e1306 100644 --- a/source/dnode/vnode/src/tsdb/tsdbDef.h +++ b/source/dnode/vnode/src/tsdb/tsdbDef.h @@ -34,7 +34,7 @@ typedef struct SFDataPtr { extern int32_t tsdbOpenFile(const char *path, STsdb *pTsdb, int32_t flag, STsdbFD **ppFD); extern void tsdbCloseFile(STsdbFD **ppFD); extern int32_t tsdbWriteFile(STsdbFD *pFD, int64_t offset, const uint8_t *pBuf, int64_t size); -extern int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size); +extern int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint); extern int32_t tsdbFsyncFile(STsdbFD *pFD); #ifdef __cplusplus diff --git a/source/dnode/vnode/src/tsdb/tsdbFS2.c b/source/dnode/vnode/src/tsdb/tsdbFS2.c index c9ecff8890..89bed6b42f 100644 --- a/source/dnode/vnode/src/tsdb/tsdbFS2.c +++ b/source/dnode/vnode/src/tsdb/tsdbFS2.c @@ -901,7 +901,7 @@ int32_t tsdbFSEditCommit(STFileSystem *fs) { if (mtime < now - tsS3UploadDelaySec) { skipMerge = true; } - } else if (s3Size(object_name) > 0) { + } else /* if (s3Size(object_name) > 0) */ { skipMerge = true; } } diff --git a/source/dnode/vnode/src/tsdb/tsdbMerge.c b/source/dnode/vnode/src/tsdb/tsdbMerge.c index 7babaa6e28..6d968d0828 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMerge.c +++ b/source/dnode/vnode/src/tsdb/tsdbMerge.c @@ -568,7 +568,7 @@ static int32_t tsdbMerge(void *arg) { if (mtime < now - tsS3UploadDelaySec) { skipMerge = true; } - } else if (s3Size(object_name) > 0) { + } else /* if (s3Size(object_name) > 0) */ { skipMerge = true; } } diff --git a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c index 8b9cae42fc..e6f419362c 100644 --- a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c +++ b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c @@ -26,7 +26,17 @@ static int32_t tsdbOpenFileImpl(STsdbFD *pFD) { if (pFD->pFD == NULL) { int errsv = errno; const char *object_name = taosDirEntryBaseName((char *)path); - long s3_size = tsS3Enabled ? s3Size(object_name) : 0; + long s3_size = 0; + if (tsS3Enabled) { + long size = s3Size(object_name); + if (size < 0) { + code = terrno = TSDB_CODE_FAILED_TO_CONNECT_S3; + goto _exit; + } + + s3_size = size; + } + if (tsS3Enabled && !strncmp(path + strlen(path) - 5, ".data", 5) && s3_size > 0) { #ifndef S3_BLOCK_CACHE s3EvictCache(path, s3_size); @@ -48,6 +58,7 @@ static int32_t tsdbOpenFileImpl(STsdbFD *pFD) { // pFD->szFile = s3_size; #endif } else { + tsdbInfo("no file: %s", path); code = TAOS_SYSTEM_ERROR(errsv); // taosMemoryFree(pFD); goto _exit; @@ -283,7 +294,7 @@ _exit: return code; } -static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) { +static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint) { int32_t code = 0; int64_t n = 0; int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage); @@ -330,7 +341,7 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64 memcpy(pBuf + n, pFD->pBuf + bOffset, nRead); n += nRead; - pgno++; + ++pgno; bOffset = 0; } @@ -339,7 +350,12 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64 uint8_t *pBlock = NULL; int64_t retrieve_offset = PAGE_OFFSET(pgno, pFD->szPage); int64_t pgnoEnd = pgno - 1 + (bOffset + size - n + szPgCont - 1) / szPgCont; - int64_t retrieve_size = (pgnoEnd - pgno + 1) * pFD->szPage; + + if (szHint > 0) { + pgnoEnd = pgno - 1 + (bOffset + szHint - n + szPgCont - 1) / szPgCont; + } + + int64_t retrieve_size = (pgnoEnd - pgno + 1) * pFD->szPage; code = s3GetObjectBlock(pFD->objName, retrieve_offset, retrieve_size, 1, &pBlock); if (code != TSDB_CODE_SUCCESS) { goto _exit; @@ -350,6 +366,10 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64 for (int i = 0; i < nPage; ++i) { tsdbCacheSetPageS3(pFD->pTsdb->pgCache, pFD, pgno, pBlock + i * pFD->szPage); + if (szHint > 0 && n >= size) { + ++pgno; + continue; + } memcpy(pFD->pBuf, pBlock + i * pFD->szPage, pFD->szPage); // check @@ -364,7 +384,7 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64 memcpy(pBuf + n, pFD->pBuf + bOffset, nRead); n += nRead; - pgno++; + ++pgno; bOffset = 0; } @@ -375,7 +395,7 @@ _exit: return code; } -int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) { +int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint) { int32_t code = 0; if (!pFD->pFD) { code = tsdbOpenFileImpl(pFD); @@ -385,7 +405,7 @@ int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) } if (pFD->s3File && tsS3BlockSize < 0) { - return tsdbReadFileS3(pFD, offset, pBuf, size); + return tsdbReadFileS3(pFD, offset, pBuf, size, szHint); } else { return tsdbReadFileImp(pFD, offset, pBuf, size); } @@ -1141,7 +1161,7 @@ int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) { if (code) goto _err; // read - code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size); + code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size, 0); if (code) goto _err; // decode @@ -1178,7 +1198,7 @@ int32_t tsdbReadSttBlk(SDataFReader *pReader, int32_t iStt, SArray *aSttBlk) { if (code) goto _err; // read - code = tsdbReadFile(pReader->aSttFD[iStt], offset, pReader->aBuf[0], size); + code = tsdbReadFile(pReader->aSttFD[iStt], offset, pReader->aBuf[0], size, 0); if (code) goto _err; // decode @@ -1211,7 +1231,7 @@ int32_t tsdbReadDataBlk(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *m if (code) goto _err; // read - code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size); + code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size, 0); if (code) goto _err; // decode @@ -1242,7 +1262,7 @@ int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aCol if (code) goto _err; // read - code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], pSmaInfo->size); + code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], pSmaInfo->size, 0); if (code) goto _err; // decode @@ -1276,7 +1296,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey); if (code) goto _err; - code = tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey); + code = tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey, 0); if (code) goto _err; SDiskDataHdr hdr; @@ -1322,7 +1342,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo code = tRealloc(&pReader->aBuf[0], hdr.szBlkCol); if (code) goto _err; - code = tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol); + code = tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol, 0); if (code) goto _err; } @@ -1366,7 +1386,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo code = tRealloc(&pReader->aBuf[1], size); if (code) goto _err; - code = tsdbReadFile(pFD, offset, pReader->aBuf[1], size); + code = tsdbReadFile(pFD, offset, pReader->aBuf[1], size, 0); if (code) goto _err; code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]); @@ -1392,7 +1412,7 @@ int32_t tsdbReadDataBlockEx(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockDat if (code) goto _err; // read - code = tsdbReadFile(pReader->pDataFD, pBlockInfo->offset, pReader->aBuf[0], pBlockInfo->szBlock); + code = tsdbReadFile(pReader->pDataFD, pBlockInfo->offset, pReader->aBuf[0], pBlockInfo->szBlock, 0); if (code) goto _err; // decmpr @@ -1444,7 +1464,7 @@ int32_t tsdbReadSttBlockEx(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk TSDB_CHECK_CODE(code, lino, _exit); // read - code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock); + code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock, 0); TSDB_CHECK_CODE(code, lino, _exit); // decmpr @@ -1700,7 +1720,7 @@ int32_t tsdbReadDelDatav1(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelDa if (code) goto _err; // read - code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size); + code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size, 0); if (code) goto _err; // // decode @@ -1740,7 +1760,7 @@ int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) { if (code) goto _err; // read - code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size); + code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size, 0); if (code) goto _err; // decode diff --git a/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c b/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c index 7c3b185e20..f26c6540df 100644 --- a/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c +++ b/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c @@ -60,7 +60,7 @@ int32_t tsdbSttFileReaderOpen(const char *fname, const SSttFileReaderConfig *con int64_t offset = config->file->size - sizeof(SSttFooter); ASSERT(offset >= TSDB_FHDR_SIZE); - code = tsdbReadFile(reader[0]->fd, offset, (uint8_t *)(reader[0]->footer), sizeof(SSttFooter)); + code = tsdbReadFile(reader[0]->fd, offset, (uint8_t *)(reader[0]->footer), sizeof(SSttFooter), 0); TSDB_CHECK_CODE(code, lino, _exit); _exit: @@ -97,7 +97,7 @@ int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray * if (!data) return TSDB_CODE_OUT_OF_MEMORY; int32_t code = - tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data, reader->footer->statisBlkPtr->size); + tsdbReadFile(reader->fd, reader->footer->statisBlkPtr->offset, data, reader->footer->statisBlkPtr->size, 0); if (code) { taosMemoryFree(data); return code; @@ -125,7 +125,7 @@ int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tom if (!data) return TSDB_CODE_OUT_OF_MEMORY; int32_t code = - tsdbReadFile(reader->fd, reader->footer->tombBlkPtr->offset, data, reader->footer->tombBlkPtr->size); + tsdbReadFile(reader->fd, reader->footer->tombBlkPtr->offset, data, reader->footer->tombBlkPtr->size, 0); if (code) { taosMemoryFree(data); return code; @@ -152,7 +152,8 @@ int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBl void *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size); if (!data) return TSDB_CODE_OUT_OF_MEMORY; - int32_t code = tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size); + int32_t code = + tsdbReadFile(reader->fd, reader->footer->sttBlkPtr->offset, data, reader->footer->sttBlkPtr->size, 0); if (code) { taosMemoryFree(data); return code; @@ -177,7 +178,7 @@ int32_t tsdbSttFileReadBlockData(SSttFileReader *reader, const SSttBlk *sttBlk, code = tRealloc(&reader->config->bufArr[0], sttBlk->bInfo.szBlock); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset, reader->config->bufArr[0], sttBlk->bInfo.szBlock); + code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset, reader->config->bufArr[0], sttBlk->bInfo.szBlock, 0); TSDB_CHECK_CODE(code, lino, _exit); code = tDecmprBlockData(reader->config->bufArr[0], sttBlk->bInfo.szBlock, bData, &reader->config->bufArr[1]); @@ -209,7 +210,7 @@ int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk * code = tRealloc(&reader->config->bufArr[0], sttBlk->bInfo.szKey); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset, reader->config->bufArr[0], sttBlk->bInfo.szKey); + code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset, reader->config->bufArr[0], sttBlk->bInfo.szKey, 0); TSDB_CHECK_CODE(code, lino, _exit); // hdr @@ -255,7 +256,7 @@ int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk * TSDB_CHECK_CODE(code, lino, _exit); code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey, reader->config->bufArr[0], - hdr->szBlkCol); + hdr->szBlkCol, 0); TSDB_CHECK_CODE(code, lino, _exit); } @@ -296,7 +297,7 @@ int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk * TSDB_CHECK_CODE(code, lino, _exit); code = tsdbReadFile(reader->fd, sttBlk->bInfo.offset + sttBlk->bInfo.szKey + hdr->szBlkCol + blockCol->offset, - reader->config->bufArr[1], size1); + reader->config->bufArr[1], size1, 0); TSDB_CHECK_CODE(code, lino, _exit); code = tsdbDecmprColData(reader->config->bufArr[1], blockCol, hdr->cmprAlg, hdr->nRow, colData, @@ -321,7 +322,7 @@ int32_t tsdbSttFileReadTombBlock(SSttFileReader *reader, const STombBlk *tombBlk code = tRealloc(&reader->config->bufArr[0], tombBlk->dp->size); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd, tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size); + code = tsdbReadFile(reader->fd, tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size, 0); if (code) TSDB_CHECK_CODE(code, lino, _exit); int64_t size = 0; @@ -352,7 +353,7 @@ int32_t tsdbSttFileReadStatisBlock(SSttFileReader *reader, const SStatisBlk *sta code = tRealloc(&reader->config->bufArr[0], statisBlk->dp->size); TSDB_CHECK_CODE(code, lino, _exit); - code = tsdbReadFile(reader->fd, statisBlk->dp->offset, reader->config->bufArr[0], statisBlk->dp->size); + code = tsdbReadFile(reader->fd, statisBlk->dp->offset, reader->config->bufArr[0], statisBlk->dp->size, 0); TSDB_CHECK_CODE(code, lino, _exit); int64_t size = 0; @@ -405,7 +406,7 @@ struct SSttFileWriter { }; static int32_t tsdbFileDoWriteSttBlockData(STsdbFD *fd, SBlockData *blockData, int8_t cmprAlg, int64_t *fileSize, - TSttBlkArray *sttBlkArray, uint8_t **bufArr, SVersionRange *range) { + TSttBlkArray *sttBlkArray, uint8_t **bufArr, SVersionRange *range) { if (blockData->nRow == 0) return 0; int32_t code = 0; diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c index f9b5e9168c..a178e1f772 100644 --- a/source/dnode/vnode/src/vnd/vnodeCommit.c +++ b/source/dnode/vnode/src/vnd/vnodeCommit.c @@ -13,9 +13,10 @@ * along with this program. If not, see . */ +#include "meta.h" +#include "sync.h" #include "vnd.h" #include "vnodeInt.h" -#include "sync.h" extern int32_t tsdbPreCommit(STsdb *pTsdb); extern int32_t tsdbCommitBegin(STsdb *pTsdb, SCommitInfo *pInfo); @@ -155,7 +156,8 @@ int vnodeShouldCommit(SVnode *pVnode, bool atExit) { taosThreadMutexLock(&pVnode->mutex); if (pVnode->inUse && diskAvail) { - needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) || (pVnode->inUse->size > 0 && atExit); + needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) || + (atExit && (pVnode->inUse->size > 0 || pVnode->pMeta->changed)); } taosThreadMutexUnlock(&pVnode->mutex); return needCommit; diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 9b79e9bd66..1f951097a4 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -594,6 +594,11 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg tqProcessTaskResetReq(pVnode->pTq, pMsg); } } break; + case TDMT_STREAM_HTASK_DROP: { + if (pVnode->restored && vnodeIsLeader(pVnode)) { + tqProcessTaskDropHTask(pVnode->pTq, pMsg); + } + } break; case TDMT_VND_ALTER_CONFIRM: needCommit = pVnode->config.hashChange; if (vnodeProcessAlterConfirmReq(pVnode, ver, pReq, len, pRsp) < 0) { diff --git a/source/libs/executor/src/dynqueryctrloperator.c b/source/libs/executor/src/dynqueryctrloperator.c index 8fc46e0239..9e7f1144f8 100755 --- a/source/libs/executor/src/dynqueryctrloperator.c +++ b/source/libs/executor/src/dynqueryctrloperator.c @@ -503,15 +503,15 @@ static void seqJoinLaunchNewRetrieveImpl(SOperatorInfo* pOperator, SSDataBlock** T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code); } - qError("dynamic post task begin"); + qDebug("%s dynamic post task begin", GET_TASKID(pOperator->pTaskInfo)); *ppRes = pOperator->pDownstream[1]->fpSet.getNextExtFn(pOperator->pDownstream[1], pParam); if (*ppRes) { pPost->isStarted = true; pStbJoin->execInfo.postBlkNum++; pStbJoin->execInfo.postBlkRows += (*ppRes)->info.rows; - qError("join res block retrieved"); + qDebug("%s join res block retrieved", GET_TASKID(pOperator->pTaskInfo)); } else { - qError("Empty join res block retrieved"); + qDebug("%s Empty join res block retrieved", GET_TASKID(pOperator->pTaskInfo)); } } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 5d4d2b7461..448c585869 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1525,6 +1525,17 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS return TSDB_CODE_SUCCESS; } +static void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) { + SExprSupp* pTbNameCalSup = &pInfo->tbnameCalSup; + blockDataCleanup(pInfo->pCreateTbRes); + if (pInfo->tbnameCalSup.numOfExprs == 0 && pInfo->tagCalSup.numOfExprs == 0) { + pBlock->info.parTbName[0] = 0; + } else { + appendCreateTableRow(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, &pInfo->tbnameCalSup, &pInfo->tagCalSup, + pBlock->info.id.groupId, pBlock, 0, pInfo->pCreateTbRes, &pInfo->stateStore); + } +} + static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) { blockDataCleanup(pDestBlock); int32_t rows = pSrcBlock->info.rows; @@ -1549,15 +1560,21 @@ static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pS for (int32_t i = 0; i < pSrcBlock->info.rows; i++) { uint64_t srcUid = srcUidData[i]; uint64_t groupId = srcGp[i]; - char tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0}; + char tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0}; if (groupId == 0) { groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], ver); } if (pInfo->tbnameCalSup.pExprInfo) { void* parTbname = NULL; - pInfo->stateStore.streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, groupId, &parTbname); - - memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN); + code = pInfo->stateStore.streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, groupId, &parTbname); + if (code != TSDB_CODE_SUCCESS) { + SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, srcUid, srcStartTsCol[i], srcStartTsCol[i], ver); + printDataBlock(pPreRes, "pre res", GET_TASKID(pInfo->pStreamScanOp->pTaskInfo)); + calBlockTbName(pInfo, pPreRes); + memcpy(varDataVal(tbname), pPreRes->info.parTbName, strlen(pPreRes->info.parTbName)); + } else { + memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN); + } varDataSetLen(tbname, strlen(varDataVal(tbname))); pInfo->stateStore.streamStateFreeVal(parTbname); } @@ -1583,17 +1600,6 @@ static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, return code; } -static void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) { - SExprSupp* pTbNameCalSup = &pInfo->tbnameCalSup; - blockDataCleanup(pInfo->pCreateTbRes); - if (pInfo->tbnameCalSup.numOfExprs == 0 && pInfo->tagCalSup.numOfExprs == 0) { - pBlock->info.parTbName[0] = 0; - } else { - appendCreateTableRow(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, &pInfo->tbnameCalSup, &pInfo->tagCalSup, - pBlock->info.id.groupId, pBlock, 0, pInfo->pCreateTbRes, &pInfo->stateStore); - } -} - void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp, void* pTbName) { SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); diff --git a/source/libs/parser/inc/parTranslater.h b/source/libs/parser/inc/parTranslater.h index e06626fb00..55ea71a368 100644 --- a/source/libs/parser/inc/parTranslater.h +++ b/source/libs/parser/inc/parTranslater.h @@ -23,6 +23,7 @@ extern "C" { #include "parToken.h" #include "parUtil.h" #include "parser.h" +#include "cmdnodes.h" typedef struct STranslateContext { SParseContext* pParseCxt; @@ -46,7 +47,9 @@ typedef struct STranslateContext { SNode* pPostRoot; } STranslateContext; +bool biRewriteToTbnameFunc(STranslateContext* pCxt, SNode** ppNode); int32_t biRewriteSelectStar(STranslateContext* pCxt, SSelectStmt* pSelect); +int32_t biCheckCreateTableTbnameCol(STranslateContext* pCxt, SCreateTableStmt* pStmt); int32_t findTable(STranslateContext* pCxt, const char* pTableAlias, STableNode** pOutput); int32_t getTargetMetaImpl(SParseContext* pParCxt, SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta, bool couldBeView); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 1f8af22241..5c30384a6b 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -263,6 +263,7 @@ static const SSysTableShowAdapter sysTableShowAdapter[] = { static int32_t translateSubquery(STranslateContext* pCxt, SNode* pNode); static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode); static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal); +static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode** pFunc); static int32_t createSimpleSelectStmtFromProjList(const char* pDb, const char* pTable, SNodeList* pProjectionList, SSelectStmt** pStmt); static int32_t createLastTsSelectStmt(char* pDb, char* pTable, STableMeta* pMeta, SNode** pQuery); @@ -1091,6 +1092,12 @@ static EDealRes translateColumnUseAlias(STranslateContext* pCxt, SColumnNode** p return DEAL_RES_CONTINUE; } +#ifndef TD_ENTERPRISE +bool biRewriteToTbnameFunc(STranslateContext* pCxt, SNode** ppNode) { + return false; +} +#endif + static EDealRes translateColumn(STranslateContext* pCxt, SColumnNode** pCol) { if (NULL == pCxt->pCurrStmt || (isSelectStmt(pCxt->pCurrStmt) && NULL == ((SSelectStmt*)pCxt->pCurrStmt)->pFromTable)) { @@ -1102,6 +1109,13 @@ static EDealRes translateColumn(STranslateContext* pCxt, SColumnNode** pCol) { return DEAL_RES_CONTINUE; } + if (pCxt->pParseCxt->biMode) { + SNode** ppNode = (SNode**)pCol; + if (biRewriteToTbnameFunc(pCxt, ppNode)) { + return translateFunction(pCxt, (SFunctionNode**)ppNode); + } + } + EDealRes res = DEAL_RES_CONTINUE; if ('\0' != (*pCol)->tableAlias[0]) { res = translateColumnWithPrefix(pCxt, pCol); @@ -2212,7 +2226,8 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode** pFunc SNode* pParam = NULL; FOREACH(pParam, (*pFunc)->pParameterList) { if (isMultiResFunc(pParam)) { - return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)pParam)->aliasName); + pCxt->errCode = TSDB_CODE_FUNC_FUNTION_PARA_NUM; + return DEAL_RES_ERROR; } } @@ -5712,6 +5727,12 @@ static int32_t checkTableDeleteMarkOption(STranslateContext* pCxt, STableOptions return code; } +#ifndef TD_ENTERPRISE +int32_t biCheckCreateTableTbnameCol(STranslateContext* pCxt, SCreateTableStmt* pStmt) { + return TSDB_CODE_SUCCESS; +} +#endif + static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt, bool createStable) { if (NULL != strchr(pStmt->tableName, '.')) { return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_IDENTIFIER_NAME, @@ -5750,7 +5771,9 @@ static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt "configured with the 'TTL' option"); } } - + if (pCxt->pParseCxt->biMode != 0 && TSDB_CODE_SUCCESS == code) { + code = biCheckCreateTableTbnameCol(pCxt, pStmt); + } return code; } diff --git a/source/libs/parser/test/parSelectTest.cpp b/source/libs/parser/test/parSelectTest.cpp index 68ded3afdd..53d97d0699 100644 --- a/source/libs/parser/test/parSelectTest.cpp +++ b/source/libs/parser/test/parSelectTest.cpp @@ -383,7 +383,7 @@ TEST_F(ParserSelectTest, semanticCheck) { run("SELECT LAST(*) + SUM(c1) FROM t1", TSDB_CODE_PAR_WRONG_VALUE_TYPE); - run("SELECT CEIL(LAST(ts, c1)) FROM t1", TSDB_CODE_PAR_WRONG_VALUE_TYPE); + run("SELECT CEIL(LAST(ts, c1)) FROM t1", TSDB_CODE_FUNC_FUNTION_PARA_NUM); // TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION run("SELECT c2 FROM t1 tt1 join t1 tt2 on COUNT(*) > 0", TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION); diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c index 6144ebd340..734d1e7d17 100644 --- a/source/libs/scalar/src/sclfunc.c +++ b/source/libs/scalar/src/sclfunc.c @@ -1457,13 +1457,29 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara } case 604800000: { /* 1w */ if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) { - timeVal = timeVal / 1000 / 604800 * 604800 * 1000; + if (ignoreTz) { + timeVal = timeVal - (timeVal + offsetFromTz(timezone, 1000)) % (((int64_t)604800) * 1000); + } else { + timeVal = timeVal / 1000 / 604800 * 604800 * 1000; + } } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) { - timeVal = timeVal / 1000000 / 604800 * 604800 * 1000000; + if (ignoreTz) { + timeVal = timeVal - (timeVal + offsetFromTz(timezone, 1000000)) % (((int64_t)604800) * 1000000); + } else { + timeVal = timeVal / 1000000 / 604800 * 604800 * 1000000; + } } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) { - timeVal = timeVal / 1000000000 / 604800 * 604800 * 1000000000; + if (ignoreTz) { + timeVal = timeVal - (timeVal + offsetFromTz(timezone, 1000000000)) % (((int64_t)604800) * 1000000000); + } else { + timeVal = timeVal / 1000000000 / 604800 * 604800 * 1000000000; + } } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) { - timeVal = timeVal * factor / factor / 604800 * 604800 * factor; + if (ignoreTz) { + timeVal = timeVal - (timeVal + offsetFromTz(timezone, 1)) % (((int64_t)604800L) * factor); + } else { + timeVal = timeVal * factor / factor / 604800 * 604800 * factor; + } } else { colDataSetNULL(pOutput->columnData, i); continue; diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h index 4800e62109..afb6b9227d 100644 --- a/source/libs/stream/inc/streamInt.h +++ b/source/libs/stream/inc/streamInt.h @@ -113,28 +113,24 @@ int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t srcTaskId, int32 int32_t streamTaskSendCheckpointReadyMsg(SStreamTask* pTask); int32_t streamTaskSendCheckpointSourceRsp(SStreamTask* pTask); int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask); - -int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks, - int32_t* blockSize); -int32_t streamQueueItemGetSize(const SStreamQueueItem* pItem); -void streamQueueItemIncSize(const SStreamQueueItem* pItem, int32_t size); -const char* streamQueueItemGetTypeStr(int32_t type); - -SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem); - -int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer, - int32_t* pLen); -int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq); -int32_t streamNotifyUpstreamContinue(SStreamTask* pTask); -int32_t streamTaskFillHistoryFinished(SStreamTask* pTask); -int32_t streamTransferStateToStreamTask(SStreamTask* pTask); - -void streamClearChkptReadyMsg(SStreamTask* pTask); - int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*); STaskId streamTaskExtractKey(const SStreamTask* pTask); void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo); +int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer, int32_t* pLen); +int32_t streamTaskFillHistoryFinished(SStreamTask* pTask); + +void streamClearChkptReadyMsg(SStreamTask* pTask); +int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks, + int32_t* blockSize); +int32_t streamQueueItemGetSize(const SStreamQueueItem* pItem); +void streamQueueItemIncSize(const SStreamQueueItem* pItem, int32_t size); +const char* streamQueueItemGetTypeStr(int32_t type); +SStreamQueueItem* streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem); + +int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq); +int32_t streamNotifyUpstreamContinue(SStreamTask* pTask); +int32_t streamTransferStateToStreamTask(SStreamTask* pTask); SStreamQueue* streamQueueOpen(int64_t cap); void streamQueueClose(SStreamQueue* pQueue, int32_t taskId); @@ -156,8 +152,8 @@ int downloadCheckpoint(char* id, char* path); int deleteCheckpoint(char* id); int deleteCheckpointFile(char* id, char* name); -int32_t onNormalTaskReady(SStreamTask* pTask); -int32_t onScanhistoryTaskReady(SStreamTask* pTask); +int32_t streamTaskOnNormalTaskReady(SStreamTask* pTask); +int32_t streamTaskOnScanhistoryTaskReady(SStreamTask* pTask); #ifdef __cplusplus } diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c index f6ec6e9fdb..bcda85e7a7 100644 --- a/source/libs/stream/src/streamData.c +++ b/source/libs/stream/src/streamData.c @@ -158,7 +158,7 @@ int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubm } // todo handle memory error -SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) { +SStreamQueueItem* streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) { terrno = 0; if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) { diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index da077c67f8..b0ba78ca6e 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -958,7 +958,9 @@ int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t upstreamTaskId, } void streamClearChkptReadyMsg(SStreamTask* pTask) { - if (pTask->pReadyMsgList == NULL) return; + if (pTask->pReadyMsgList == NULL) { + return; + } for (int i = 0; i < taosArrayGetSize(pTask->pReadyMsgList); i++) { SStreamChkptReadyInfo* pInfo = taosArrayGet(pTask->pReadyMsgList, i); diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 43875319b7..8b14846414 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -401,7 +401,7 @@ int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) { // 5. save to disk pStreamTask->status.taskStatus = streamTaskGetStatus(pStreamTask, NULL); - // 6. pause allowed. + // 6. add empty delete block if ((pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE) && taosQueueEmpty(pStreamTask->inputq.queue->pQueue)) { SStreamRefDataBlock* pItem = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0); diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c index d19dfc13bf..d1610362f9 100644 --- a/source/libs/stream/src/streamQueue.c +++ b/source/libs/stream/src/streamQueue.c @@ -221,7 +221,7 @@ int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu *pInput = qItem; } else { // merge current block failed, let's handle the already merged blocks. - void* newRet = streamMergeQueueItem(*pInput, qItem); + void* newRet = streamQueueMergeQueueItem(*pInput, qItem); if (newRet == NULL) { if (terrno != 0) { stError("s-task:%s failed to merge blocks from inputQ, numOfBlocks:%d, code:%s", id, *numOfBlocks, diff --git a/source/libs/stream/src/streamStart.c b/source/libs/stream/src/streamStart.c index e5063e7d52..71f1e9b45b 100644 --- a/source/libs/stream/src/streamStart.c +++ b/source/libs/stream/src/streamStart.c @@ -323,7 +323,7 @@ int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_ } } -int32_t onNormalTaskReady(SStreamTask* pTask) { +int32_t streamTaskOnNormalTaskReady(SStreamTask* pTask) { const char* id = pTask->id.idStr; streamTaskSetReady(pTask); @@ -348,7 +348,7 @@ int32_t onNormalTaskReady(SStreamTask* pTask) { return TSDB_CODE_SUCCESS; } -int32_t onScanhistoryTaskReady(SStreamTask* pTask) { +int32_t streamTaskOnScanhistoryTaskReady(SStreamTask* pTask) { const char* id = pTask->id.idStr; // set the state to be ready diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 69a8d1309a..2f0bfc8dbc 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -451,7 +451,7 @@ int32_t streamTaskInit(SStreamTask* pTask, SStreamMeta* pMeta, SMsgCb* pMsgCb, i // 2MiB per second for sink task // 50 times sink operator per second - streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 50, 50, tsSinkDataRate, pTask->id.idStr); + streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 35, 35, tsSinkDataRate, pTask->id.idStr); TdThreadMutexAttr attr = {0}; int code = taosThreadMutexAttrInit(&attr); diff --git a/source/libs/stream/src/streamTaskSm.c b/source/libs/stream/src/streamTaskSm.c index 1c951e1452..cac3766893 100644 --- a/source/libs/stream/src/streamTaskSm.c +++ b/source/libs/stream/src/streamTaskSm.c @@ -457,11 +457,11 @@ void doInitStateTransferTable(void) { streamTaskSMTrans = taosArrayInit(8, sizeof(STaskStateTrans)); // initialization event handle - STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT, streamTaskInitStatus, onNormalTaskReady, false, false); + STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT, streamTaskInitStatus, streamTaskOnNormalTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); - trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false); + trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST, streamTaskInitStatus, streamTaskOnScanhistoryTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); - trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false); + trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST, streamTaskInitStatus, streamTaskOnScanhistoryTaskReady, false, false); taosArrayPush(streamTaskSMTrans, &trans); // scan-history related event diff --git a/source/util/src/tcompression.c b/source/util/src/tcompression.c index 06b82f3ba1..09f50d0f6d 100644 --- a/source/util/src/tcompression.c +++ b/source/util/src/tcompression.c @@ -280,8 +280,7 @@ int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, cha #endif } -/* ----------------------------------------------Bool Compression - * ---------------------------------------------- */ +/* ----------------------------------------------Bool Compression ---------------------------------------------- */ // TODO: You can also implement it using RLE method. int32_t tsCompressBoolImp(const char *const input, const int32_t nelements, char *const output) { int32_t pos = -1; @@ -387,8 +386,7 @@ int32_t tsDecompressBoolRLEImp(const char *const input, const int32_t nelements, } #endif -/* ----------------------------------------------String Compression - * ---------------------------------------------- */ +/* ----------------------------------------------String Compression ---------------------------------------------- */ // Note: the size of the output must be larger than input_size + 1 and // LZ4_compressBound(size) + 1; // >= max(input_size, LZ4_compressBound(input_size)) + 1; @@ -430,8 +428,7 @@ int32_t tsDecompressStringImp(const char *const input, int32_t compressedSize, c } } -/* --------------------------------------------Timestamp Compression - * ---------------------------------------------- */ +/* --------------------------------------------Timestamp Compression ---------------------------------------------- */ // TODO: Take care here, we assumes little endian encoding. int32_t tsCompressTimestampImp(const char *const input, const int32_t nelements, char *const output) { int32_t _pos = 1; @@ -751,8 +748,7 @@ int32_t tsDecompressDoubleImp(const char *const input, const int32_t nelements, return nelements * DOUBLE_BYTES; } -/* --------------------------------------------Float Compression - * ---------------------------------------------- */ +/* --------------------------------------------Float Compression ---------------------------------------------- */ void encodeFloatValue(uint32_t diff, uint8_t flag, char *const output, int32_t *const pos) { uint8_t nbytes = (flag & INT8MASK(3)) + 1; int32_t nshift = (FLOAT_BYTES * BITS_PER_BYTE - nbytes * BITS_PER_BYTE) * (flag >> 3); diff --git a/source/util/src/tdecompress.c b/source/util/src/tdecompress.c index f32a4014d6..68841941db 100644 --- a/source/util/src/tdecompress.c +++ b/source/util/src/tdecompress.c @@ -50,7 +50,7 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, const char *ip = input + 1; int32_t count = 0; int32_t _pos = 0; - int64_t prev_value = 0; + int64_t prevValue = 0; #if __AVX2__ while (1) { @@ -80,13 +80,13 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, if (selector == 0 || selector == 1) { if (tsSIMDEnable && tsAVX2Enable) { for (int32_t i = 0; i < batch; ++i) { - __m256i prev = _mm256_set1_epi64x(prev_value); + __m256i prev = _mm256_set1_epi64x(prevValue); _mm256_storeu_si256((__m256i *)&p[_pos], prev); _pos += 4; } for (int32_t i = 0; i < remain; ++i) { - p[_pos++] = prev_value; + p[_pos++] = prevValue; } } else if (tsSIMDEnable && tsAVX512Enable) { #if __AVX512F__ @@ -94,7 +94,7 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, #endif } else { // alternative implementation without SIMD instructions. for (int32_t i = 0; i < elems && count < nelements; i++, count++) { - p[_pos++] = prev_value; + p[_pos++] = prevValue; v += bit; } } @@ -118,16 +118,16 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, __m256i delta = _mm256_xor_si256(_mm256_srli_epi64(zigzagVal, 1), signmask); // calculate the cumulative sum (prefix sum) for each number - // decode[0] = prev_value + final[0] - // decode[1] = decode[0] + final[1] -----> prev_value + final[0] + final[1] - // decode[2] = decode[1] + final[2] -----> prev_value + final[0] + final[1] + final[2] - // decode[3] = decode[2] + final[3] -----> prev_value + final[0] + final[1] + final[2] + final[3] + // decode[0] = prevValue + final[0] + // decode[1] = decode[0] + final[1] -----> prevValue + final[0] + final[1] + // decode[2] = decode[1] + final[2] -----> prevValue + final[0] + final[1] + final[2] + // decode[3] = decode[2] + final[3] -----> prevValue + final[0] + final[1] + final[2] + final[3] // 1, 2, 3, 4 //+ 0, 1, 0, 3 // 1, 3, 3, 7 // shift and add for the first round - __m128i prev = _mm_set1_epi64x(prev_value); + __m128i prev = _mm_set1_epi64x(prevValue); __m256i x = _mm256_slli_si256(delta, 8); delta = _mm256_add_epi64(delta, x); @@ -148,16 +148,16 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, _mm_storeu_si128((__m128i *)&p[_pos + 2], secPart); shiftBits = _mm256_add_epi64(shiftBits, inc); - prev_value = p[_pos + 3]; + prevValue = p[_pos + 3]; _pos += 4; } // handle the remain value for (int32_t i = 0; i < remain; i++) { zigzag_value = ((w >> (v + (batch * bit * 4))) & mask); - prev_value += ZIGZAG_DECODE(int64_t, zigzag_value); + prevValue += ZIGZAG_DECODE(int64_t, zigzag_value); - p[_pos++] = prev_value; + p[_pos++] = prevValue; v += bit; } } else if (tsSIMDEnable && tsAVX512Enable) { @@ -167,9 +167,9 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, } else { // alternative implementation without SIMD instructions. for (int32_t i = 0; i < elems && count < nelements; i++, count++) { zigzag_value = ((w >> v) & mask); - prev_value += ZIGZAG_DECODE(int64_t, zigzag_value); + prevValue += ZIGZAG_DECODE(int64_t, zigzag_value); - p[_pos++] = prev_value; + p[_pos++] = prevValue; v += bit; } } @@ -180,14 +180,14 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, if (selector == 0 || selector == 1) { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { - p[_pos++] = (int32_t)prev_value; + p[_pos++] = (int32_t)prevValue; } } else { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { zigzag_value = ((w >> v) & mask); - prev_value += ZIGZAG_DECODE(int64_t, zigzag_value); + prevValue += ZIGZAG_DECODE(int64_t, zigzag_value); - p[_pos++] = (int32_t)prev_value; + p[_pos++] = (int32_t)prevValue; v += bit; } } @@ -197,14 +197,14 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, if (selector == 0 || selector == 1) { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { - p[_pos++] = (int16_t)prev_value; + p[_pos++] = (int16_t)prevValue; } } else { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { zigzag_value = ((w >> v) & mask); - prev_value += ZIGZAG_DECODE(int64_t, zigzag_value); + prevValue += ZIGZAG_DECODE(int64_t, zigzag_value); - p[_pos++] = (int16_t)prev_value; + p[_pos++] = (int16_t)prevValue; v += bit; } } @@ -215,14 +215,14 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, if (selector == 0 || selector == 1) { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { - p[_pos++] = (int8_t)prev_value; + p[_pos++] = (int8_t)prevValue; } } else { for (int32_t i = 0; i < elems && count < nelements; i++, count++) { zigzag_value = ((w >> v) & mask); - prev_value += ZIGZAG_DECODE(int64_t, zigzag_value); + prevValue += ZIGZAG_DECODE(int64_t, zigzag_value); - p[_pos++] = (int8_t)prev_value; + p[_pos++] = (int8_t)prevValue; v += bit; } } @@ -246,6 +246,77 @@ int32_t tsDecompressFloatImplAvx512(const char *const input, const int32_t nelem // todo add later int32_t tsDecompressFloatImplAvx2(const char *const input, const int32_t nelements, char *const output) { #if __AVX2__ +#endif + return 0; +} + +int32_t tsDecompressTimestampAvx2(const char* const input, const int32_t nelements, char *const output, bool bigEndian) { + int64_t *ostream = (int64_t *)output; + int32_t ipos = 1, opos = 0; + int8_t nbytes = 0; + + int64_t prevValue = 0; + int64_t prevDelta = 0; + + int64_t deltaOfDelta = 0; + int32_t longBytes = LONG_BYTES; + +#if __AVX2__ + + int32_t batch = nelements >> 2; + int32_t remainder = nelements & 0x1; + + while (1) { + uint8_t flags = input[ipos++]; + + // Decode dd1 + uint64_t dd1 = 0; + nbytes = flags & INT8MASK(4); + + if (nbytes == 0) { + deltaOfDelta = 0; + } else { + if (bigEndian) { + memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes); + } else { + memcpy(&dd1, input + ipos, nbytes); + } + deltaOfDelta = ZIGZAG_DECODE(int64_t, dd1); + } + + ipos += nbytes; + prevDelta += deltaOfDelta; + prevValue += prevDelta; + ostream[opos++] = prevValue; + + if (opos == nelements) { + return nelements * longBytes; + } + + // Decode dd2 + uint64_t dd2 = 0; + nbytes = (flags >> 4) & INT8MASK(4); + if (nbytes == 0) { + deltaOfDelta = 0; + } else { + if (bigEndian) { + memcpy(((char *)(&dd2)) + longBytes - nbytes, input + ipos, nbytes); + } else { + memcpy(&dd2, input + ipos, nbytes); + } + // zigzag_decoding + deltaOfDelta = ZIGZAG_DECODE(int64_t, dd2); + } + + ipos += nbytes; + prevDelta += deltaOfDelta; + prevValue += prevDelta; + ostream[opos++] = prevValue; + + if (opos == nelements) { + return nelements * longBytes; + } + } #endif return 0; } \ No newline at end of file diff --git a/source/util/src/terror.c b/source/util/src/terror.c index f310db53ef..8847b7d894 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -101,6 +101,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_APP_IS_STOPPING, "Database is closing d TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_DATA_FMT, "Invalid data format") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_CFG_VALUE, "Invalid configuration value") TAOS_DEFINE_ERROR(TSDB_CODE_IP_NOT_IN_WHITE_LIST, "Not allowed to connect") +TAOS_DEFINE_ERROR(TSDB_CODE_FAILED_TO_CONNECT_S3, "Failed to connect to s3 server") //client TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_OPERATION, "Invalid operation") diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index ca1cc704ae..8cb23c081f 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -299,6 +299,8 @@ e ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionUS.py ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionNS.py ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_ts4219.py +,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_ts4295.py +,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_td27388.py ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show_tag_index.py ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/information_schema.py @@ -1068,6 +1070,7 @@ e ,,y,script,./test.sh -f tsim/query/show_db_table_kind.sim ,,y,script,./test.sh -f tsim/query/bi_star_table.sim ,,y,script,./test.sh -f tsim/query/bi_tag_scan.sim +,,y,script,./test.sh -f tsim/query/bi_tbname_col.sim ,,y,script,./test.sh -f tsim/query/tag_scan.sim ,,y,script,./test.sh -f tsim/query/nullColSma.sim ,,y,script,./test.sh -f tsim/query/bug3398.sim diff --git a/tests/script/tsim/query/bi_tbname_col.sim b/tests/script/tsim/query/bi_tbname_col.sim new file mode 100644 index 0000000000..d351a92539 --- /dev/null +++ b/tests/script/tsim/query/bi_tbname_col.sim @@ -0,0 +1,36 @@ + +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/exec.sh -n dnode1 -s start +sql connect + +sql drop database if exists db1; +sql create database db1 vgroups 3; +sql create database db1; +sql use db1; +sql create stable sta (ts timestamp, f1 int, f2 binary(200)) tags(t1 int, t2 int, t3 int); +sql create stable stb (ts timestamp, f1 int, f2 binary(200)) tags(t1 int, t2 int, t3 int); +sql create table tba1 using sta tags(1, 1, 1); +sql create table tba2 using sta tags(2, 2, 2); +sql insert into tba1 values(now, 1, "1")(now+3s, 3, "3")(now+5s, 5, "5"); +sql insert into tba2 values(now + 1s, 2, "2")(now+2s, 2, "2")(now+4s, 4, "4"); +sql create table tbn1 (ts timestamp, f1 int); + +set_bi_mode 1 + +sql select `tbname`, f1, f2 from sta order by ts +print $rows +print $data00 $data01 $data02 $data10 $data11 $data12 +if $rows != 6 then + return -1 +endi +if $data00 != @tba1@ then + return -1 +endi +if $data10 != @tba2@ then + return -1 +endi + +sql_error create table stc(ts timestamp, `tbname` binary(200)); +sql_error create table std(ts timestamp, f1 int) tags(`tbname` binary(200)); +system sh/exec.sh -n dnode1 -s stop -x SIGINT diff --git a/tests/system-test/1-insert/alter_stable.py b/tests/system-test/1-insert/alter_stable.py index 52f185a868..40d7f04ceb 100644 --- a/tests/system-test/1-insert/alter_stable.py +++ b/tests/system-test/1-insert/alter_stable.py @@ -13,6 +13,7 @@ import random import string +import threading from util.log import * from util.cases import * from util.sql import * @@ -25,10 +26,24 @@ class TDTestCase: tdLog.debug("start to execute %s" % __file__) tdSql.init(conn.cursor()) self.setsql = TDSetSql() + self.fname = __file__ + '.tmp.sql' + self.dbname = 'db1' self.ntbname = 'ntb' self.stbname = 'stb' + self.stbnum = 10 + self.ntbnum = 10 + self.colnum = 52 + self.tagnum = 15 + self.collen = 320 + self.colnum_modify = 40 + self.tagnum_modify = 40 + self.collen_old_modify = 160 + self.collen_new_modify = 455 + self.taglen_old_modify = 80 + self.taglen_new_modify = 155 self.binary_length = 20 # the length of binary for column_dict self.nchar_length = 20 # the length of nchar for column_dict + self.threadnum = 2 self.column_dict = { 'ts' : 'timestamp', 'col1': 'tinyint', @@ -183,9 +198,114 @@ class TDTestCase: tdLog.info(res) assert(res[1][2] == 39001) + def prepareAlterEnv(self): + tdSql.execute(f'drop database if exists {self.dbname}') + tdSql.execute(f'create database if not exists {self.dbname} vgroups 2') + tdSql.execute(f'use {self.dbname}') + + def destroyAlterEnv(self): + tdSql.execute(f'drop database if exists {self.dbname}') + + def alterTableTask(self, i): + os.system(f'taos -f {self.fname}.{i};') + + def executeAlterTable(self, opt): + threads = [] + for i in range(self.threadnum): + thread = threading.Thread(target=self.alterTableTask, args=(i,)) + threads.append(thread) + thread.start() + for i in range(self.threadnum): + threads[i].join() + + def checkAlterTable(self, opt): + if opt in ["stb_add_col", "stb_add_tag"]: + for i in range(self.stbnum): + tdSql.execute(f'desc {self.stbname}_{i}') + elif opt in ["stb_modify_col", "stb_modify_tag"]: + for i in range(self.stbnum): + tdSql.execute(f'desc {self.stbname}_{i}') + elif opt in ["ntb_add_col", "ntb_modify_col"]: + for i in range(self.ntbnum): + tdSql.execute(f'desc {self.ntbname}_{i}') + + def destroyAlterTable(self): + for i in range(self.threadnum): + if os.path.isfile(f'{self.fname}.{i}'): + os.remove(f'{self.fname}.{i}') + + def prepareAlterTable(self, opt): + self.destroyAlterTable() + lines = [f'use {self.dbname};\n'] + if opt in ["stb_add_col", "stb_add_tag"]: + for i in range(self.stbnum): + tdSql.execute(f'create table if not exists {self.stbname}_{i} (ts timestamp, c_0 NCHAR({self.collen})) tags(t0 nchar({self.collen}));') + for i in range(self.stbnum): + if opt == 'stb_add_col': + for c in range(1, self.colnum): + lines.append(f'alter table {self.stbname}_{i} add column c_{c} NCHAR({self.collen});\n') + else: + for c in range(1, self.tagnum): + lines.append(f'alter table {self.stbname}_{i} add tag t_{c} NCHAR({self.collen});\n') + elif opt in ["stb_modify_col", "stb_modify_tag"]: + for i in range(self.stbnum): + createTbSql = f'CREATE table if not exists {self.stbname}_{i} (ts timestamp' + for j in range(self.colnum_modify): + createTbSql += f',c_{j} NCHAR({self.collen_old_modify})' + createTbSql += f') tags(t_0 NCHAR({self.taglen_old_modify})' + for k in range(1,self.tagnum_modify): + createTbSql += f',t_{k} NCHAR({self.taglen_old_modify})' + createTbSql += f');' + tdLog.info(createTbSql) + tdSql.execute(createTbSql) + for i in range(self.stbnum): + if opt == 'stb_modify_col': + for c in range(self.colnum_modify): + lines.append(f'alter table {self.stbname}_{i} modify column c_{c} NCHAR({self.collen_new_modify});\n') + else: + for c in range(self.tagnum_modify): + lines.append(f'alter table {self.stbname}_{i} modify tag t_{c} NCHAR({self.taglen_new_modify});\n') + elif opt in ['ntb_add_col']: + for i in range(self.ntbnum): + tdSql.execute(f'create table if not exists {self.ntbname}_{i} (ts timestamp, c_0 NCHAR({self.collen}));') + for i in range(self.ntbnum): + for c in range(1, self.colnum): + lines.append(f'alter table {self.ntbname}_{i} add column c_{c} NCHAR({self.collen});\n') + elif opt in ['ntb_modify_col']: + for i in range(self.ntbnum): + createTbSql = f'CREATE table if not exists {self.ntbname}_{i} (ts timestamp' + for j in range(self.colnum_modify): + createTbSql += f',c_{j} NCHAR({self.collen_old_modify})' + createTbSql += f');' + tdLog.info(createTbSql) + tdSql.execute(createTbSql) + for i in range(self.ntbnum): + for c in range(self.colnum_modify): + lines.append(f'alter table {self.ntbname}_{i} modify column c_{c} NCHAR({self.collen_new_modify});\n') + # generate sql file + with open(f'{self.fname}.0', "a") as f: + f.writelines(lines) + # clone sql file in case of race condition + for i in range(1, self.threadnum): + shutil.copy(f'{self.fname}.0', f'{self.fname}.{i}') + + def alter_stable_multi_client_check(self): + """Check alter stable/ntable var type column/tag(PI-23) + """ + alter_table_check_type = ["stb_add_col", "stb_add_tag", "stb_modify_col", "stb_modify_tag", "ntb_add_col", "ntb_modify_col"] + + for opt in alter_table_check_type: + self.prepareAlterEnv() + self.prepareAlterTable(opt) + self.executeAlterTable(opt) + self.checkAlterTable(opt) + self.destroyAlterTable() + self.destroyAlterEnv() + def run(self): self.alter_stable_check() self.alter_stable_column_varchar_39001() + self.alter_stable_multi_client_check() def stop(self): tdSql.close() tdLog.success("%s successfully executed" % __file__) diff --git a/tests/system-test/1-insert/test_td27388.py b/tests/system-test/1-insert/test_td27388.py new file mode 100644 index 0000000000..7b49a63dbb --- /dev/null +++ b/tests/system-test/1-insert/test_td27388.py @@ -0,0 +1,97 @@ +import random +import string +from util.log import * +from util.cases import * +from util.sql import * +from util.sqlset import * +from util import constant +from util.common import * + + +class TDTestCase: + """Verify the insert with format exception for task TD-27388 + """ + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor()) + self.dbname = 'db' + self.stbname = 'st' + tdSql.execute("create database {};".format(self.dbname)) + tdSql.execute("use {};".format(self.dbname)) + tdSql.execute("create table st (ts timestamp, col1 int, col2 varchar(64)) tags (t1 int, t2 varchar(32));") + + def test_half_quotes(self): + sql_list = [ + "insert into t1 using st tags(1, 'tag1) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, tag1') values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, test msg');", + "insert into t1 using st tags(1, 'tag1' values(now, 1, test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg)';", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2) values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, tag2') values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1) values(now, 1, 'test msg') t2 using st tags(2, 'tag2) values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg) t2 using st tags(2, 'tag2') values(now, 2, test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2') values(now, 2, 'test msg);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2') values(now, 2, 'test msg);" + ] + for sql in sql_list: + tdLog.debug("execute harlf quotes sql: %s" % sql) + tdSql.error(sql) + + def test_esc(self): + sql_list = [ + "insert into t1 using st tags(1, 'tag1\\') values(now, 1, 'test msg');", + "insert into t1 using st tags(1, \\'tag1') values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg\\');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, \\'test msg');", + "insert into t1 using st tags(1, \\'tag1\\') values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, \\'test msg\\');", + "insert into t1 using st tags(1, \\'tag1\\') values(now, 1, \\'test msg\\');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2\\') values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, \\'tag2') values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2') values(now, 2, \\'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2') values(now, 2, 'test msg\\');", + "insert into t1 using st tags(1, \\'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2\\') values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2') values(now, 2, \\'test msg\\');" + ] + for sql in sql_list: + tdLog.debug("execute escape character sql: %s" % sql) + tdSql.error(sql) + + def test_specific_character(self): + sql_list = [ + "insert into t1 using st tags(1, 'tag1$) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1,) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1'') values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1() values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1*) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1+) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1,) values(now, 1, 'test msg');", + "isnert into t1 using st tags(1, 'tag1-) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1.) values(now, 1, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg$);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg,);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg+%+-.);", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2$) values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2,) values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2'') values(now, 2, 'test msg');", + "insert into t1 using st tags(1, 'tag1') values(now, 1, 'test msg') t2 using st tags(2, 'tag2() values(now, 2, 'test msg');" + ] + for sql in sql_list: + tdLog.debug("execute specific character sql: %s" % sql) + tdSql.error(sql) + + def run(self): + self.test_half_quotes() + self.test_esc() + self.test_specific_character() + + def stop(self): + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) diff --git a/tests/system-test/1-insert/test_ts4295.py b/tests/system-test/1-insert/test_ts4295.py new file mode 100644 index 0000000000..89e445f3c1 --- /dev/null +++ b/tests/system-test/1-insert/test_ts4295.py @@ -0,0 +1,49 @@ +import os +import sys +from util.log import * +from util.cases import * +from util.sql import * +from util.dnodes import tdDnodes +from math import inf +import taos + +class TDTestCase: + """Verify inserting varbinary type data of ts-4295 + """ + def init(self, conn, logSql, replicaVer=1): + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), True) + self.conn = conn + self.db_name = "db" + self.stable_name = "st" + + def run(self): + tdSql.execute("create database if not exists %s" % self.db_name) + tdSql.execute("use %s" % self.db_name) + # create super table + tdSql.execute("create table %s (ts timestamp, c1 varbinary(32)) tags (t1 int)" % self.stable_name) + # create child table + child_table_list = [] + for i in range(10): + child_table_name = "ct_" + str(i+1) + child_table_list.append(child_table_name) + tdSql.execute("create table %s using st tags(%s);" % (child_table_name, str(i+1))) + tdLog.info("create table %s successfully" % child_table_name) + # insert data + for i in range(100): + sql = "insert into table_name values" + for j in range(10000): + sql += "(now+%ss, '0x7661726331')," % str(j+1) + for child_table in child_table_list: + tdSql.execute(sql.replace("table_name", child_table)) + tdLog.info("Insert data into %s successfully" % child_table) + tdLog.info("Insert data round %s successfully" % str(i+1)) + tdSql.execute("flush database %s" % self.db_name) + + def stop(self): + tdSql.execute("drop database if exists %s" % self.db_name) + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) diff --git a/tests/system-test/2-query/timetruncate.py b/tests/system-test/2-query/timetruncate.py index a59180c2b1..09bdfcef63 100644 --- a/tests/system-test/2-query/timetruncate.py +++ b/tests/system-test/2-query/timetruncate.py @@ -58,7 +58,11 @@ class TDTestCase: elif unit.lower() == '1w': for i in range(len(self.ts_str)): ts_result = self.get_time.get_ms_timestamp(str(tdSql.queryResult[i][0])) - tdSql.checkEqual(ts_result,int(date_time[i]/1000/60/60/24/7)*7*24*60*60*1000) + if ignore_tz == 0: + tdSql.checkEqual(ts_result,int(date_time[i]/1000/60/60/24/7)*7*24*60*60*1000) + else: + # assuming the client timezone is UTC+0800 + tdSql.checkEqual(ts_result,int(date_time[i] - (date_time[i] + 8 * 3600 * 1000) % (86400 * 7 * 1000))) def check_us_timestamp(self,unit,date_time, ignore_tz): if unit.lower() == '1u': @@ -92,7 +96,11 @@ class TDTestCase: elif unit.lower() == '1w': for i in range(len(self.ts_str)): ts_result = self.get_time.get_us_timestamp(str(tdSql.queryResult[i][0])) - tdSql.checkEqual(ts_result,int(date_time[i]/1000/1000/60/60/24/7)*7*24*60*60*1000*1000) + if ignore_tz == 0: + tdSql.checkEqual(ts_result,int(date_time[i]/1000/1000/60/60/24/7)*7*24*60*60*1000*1000) + else: + # assuming the client timezone is UTC+0800 + tdSql.checkEqual(ts_result,int(date_time[i] - (date_time[i] + 8 * 3600 * 1000000) % (86400 * 7 * 1000000))) def check_ns_timestamp(self,unit,date_time, ignore_tz): if unit.lower() == '1b': @@ -130,7 +138,11 @@ class TDTestCase: elif unit.lower() == '1w': for i in range(len(self.ts_str)): if self.rest_tag != 'rest': - tdSql.checkEqual(tdSql.queryResult[i][0],int(date_time[i]*1000/1000/1000/1000/1000/60/60/24/7)*7*24*60*60*1000*1000*1000) + if ignore_tz == 0: + tdSql.checkEqual(tdSql.queryResult[i][0],int(date_time[i]*1000/1000/1000/1000/1000/60/60/24/7)*7*24*60*60*1000*1000*1000) + else: + # assuming the client timezone is UTC+0800 + tdSql.checkEqual(tdSql.queryResult[i][0],int(date_time[i] - (date_time[i] + 8 * 3600 * 1000000000) % (86400 * 7 * 1000000000))) def check_tb_type(self,unit,tb_type,ignore_tz): if tb_type.lower() == 'ntb': diff --git a/utils/tsim/src/simExe.c b/utils/tsim/src/simExe.c index ac17fad36b..394b168b08 100644 --- a/utils/tsim/src/simExe.c +++ b/utils/tsim/src/simExe.c @@ -693,8 +693,8 @@ bool simExecuteNativeSqlCommand(SScript *script, char *rest, bool isSlow) { ret = 0; break; } else if (ret != 0) { - simDebug("script:%s, taos:%p, %s failed, ret:%d:%s, error:%s", script->fileName, script->taos, rest, ret & 0XFFFF, - tstrerror(ret), taos_errstr(pSql)); + simDebug("script:%s, taos:%p, %s failed, ret:%d:%s", script->fileName, script->taos, rest, ret & 0XFFFF, + tstrerror(ret)); if (line->errorJump == SQL_JUMP_TRUE) { script->linePos = line->jump;