diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index 19a6407b77..8d63306a88 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -78,7 +78,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t v * @param SReadHandle * @return */ -qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, SSchemaWrapper** pSchema); +qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, uint64_t id); /** * set the task Id, usually used by message queue process @@ -89,6 +89,7 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int3 void qSetTaskId(qTaskInfo_t tinfo, uint64_t taskId, uint64_t queryId); int32_t qSetStreamOpOpen(qTaskInfo_t tinfo); + /** * Set multiple input data blocks for the stream scan. * @param tinfo diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 111ca28cdc..a16ddce0aa 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -535,10 +535,14 @@ static int32_t doSendCommitMsg(tmq_t* tmq, SMqClientVg* pVg, const char* pTopicN atomic_add_fetch_32(&pParamSet->totalRspNum, 1); SEp* pEp = GET_ACTIVE_EP(&pVg->epSet); - tscDebug("consumer:0x%" PRIx64 " topic:%s on vgId:%d send offset:%" PRId64 " prev:%" PRId64 - ", ep:%s:%d, ordinal:%d/%d, req:0x%" PRIx64, - tmq->consumerId, pOffset->subKey, pVg->vgId, pOffset->val.version, pVg->committedOffset.version, pEp->fqdn, - pEp->port, index + 1, totalVgroups, pMsgSendInfo->requestId); + char offsetBuf[80] = {0}; + tFormatOffset(offsetBuf, tListLen(offsetBuf), &pOffset->val); + + char commitBuf[80] = {0}; + tFormatOffset(commitBuf, tListLen(commitBuf), &pVg->committedOffset); + tscDebug("consumer:0x%" PRIx64 " topic:%s on vgId:%d send offset:%s prev:%s, ep:%s:%d, ordinal:%d/%d, req:0x%" PRIx64, + tmq->consumerId, pOffset->subKey, pVg->vgId, offsetBuf, commitBuf, pEp->fqdn, pEp->port, index + 1, + totalVgroups, pMsgSendInfo->requestId); int64_t transporterId = 0; asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, pMsgSendInfo); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 4db53c1627..d4bdd633e9 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -399,8 +399,8 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand char formatBuf[80]; tFormatOffset(formatBuf, 80, pOffsetVal); - tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue.", - consumerId, pHandle->subKey, vgId, formatBuf); + tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue. reqId:0x%"PRIx64, + consumerId, pHandle->subKey, vgId, formatBuf, pRequest->reqId); return 0; } else { // no poll occurs in this vnode for this topic, let's seek to the right offset value. @@ -799,7 +799,6 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg STqHandle tqHandle = {0}; pHandle = &tqHandle; - /*taosInitRWLatch(&pExec->lock);*/ uint64_t oldConsumerId = pHandle->consumerId; memcpy(pHandle->subKey, req.subKey, TSDB_SUBSCRIBE_KEY_LEN); @@ -834,7 +833,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg req.qmsg = NULL; pHandle->execHandle.task = - qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, vgId, &pHandle->execHandle.numOfCols, NULL); + qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, vgId, &pHandle->execHandle.numOfCols, req.newConsumerId); void* scanner = NULL; qExtractStreamScanner(pHandle->execHandle.task, &scanner); pHandle->execHandle.pExecReader = qExtractReaderFromStreamScanner(scanner); @@ -847,7 +846,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg buildSnapContext(handle.meta, handle.version, 0, pHandle->execHandle.subType, pHandle->fetchMeta, (SSnapContext**)(&handle.sContext)); - pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, vgId, NULL, NULL); + pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, vgId, NULL, req.newConsumerId); } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) { pHandle->pWalReader = walOpenReader(pVnode->pWal, NULL); pHandle->execHandle.execTb.suid = req.suid; @@ -865,7 +864,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg buildSnapContext(handle.meta, handle.version, req.suid, pHandle->execHandle.subType, pHandle->fetchMeta, (SSnapContext**)(&handle.sContext)); - pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, vgId, NULL, NULL); + pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, vgId, NULL, req.newConsumerId); } taosHashPut(pTq->pHandle, req.subKey, strlen(req.subKey), pHandle, sizeof(STqHandle)); diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index b6bca1e4ca..a273f2edec 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -307,7 +307,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { handle.execHandle.task = - qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, vgId, &handle.execHandle.numOfCols, NULL); + qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, vgId, &handle.execHandle.numOfCols, 0); if (handle.execHandle.task == NULL) { tqError("cannot create exec task for %s", handle.subKey); code = -1; @@ -332,7 +332,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { buildSnapContext(reader.meta, reader.version, 0, handle.execHandle.subType, handle.fetchMeta, (SSnapContext**)(&reader.sContext)); - handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, NULL); + handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, 0); } else if (handle.execHandle.subType == TOPIC_SUB_TYPE__TABLE) { handle.pWalReader = walOpenReader(pTq->pVnode->pWal, NULL); @@ -341,7 +341,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { tqDebug("vgId:%d, tq try to get all ctb, suid:%" PRId64, pTq->pVnode->config.vgId, handle.execHandle.execTb.suid); for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) { int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); - tqDebug("vgId:%d, idx %d, uid:%" PRId64, TD_VID(pTq->pVnode), i, tbUid); + tqDebug("vgId:%d, idx %d, uid:%" PRId64, vgId, i, tbUid); } handle.execHandle.pExecReader = tqOpenReader(pTq->pVnode); tqReaderSetTbUidList(handle.execHandle.pExecReader, tbUidList); @@ -349,9 +349,9 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { buildSnapContext(reader.meta, reader.version, handle.execHandle.execTb.suid, handle.execHandle.subType, handle.fetchMeta, (SSnapContext**)(&reader.sContext)); - handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, NULL); + handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, 0); } - tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle.subKey, handle.consumerId, TD_VID(pTq->pVnode)); + tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle.subKey, handle.consumerId, vgId); taosHashPut(pTq->pHandle, pKey, kLen, &handle, sizeof(STqHandle)); } diff --git a/source/dnode/vnode/src/tq/tqScan.c b/source/dnode/vnode/src/tq/tqScan.c index 1a166d326f..85a62c4dd1 100644 --- a/source/dnode/vnode/src/tq/tqScan.c +++ b/source/dnode/vnode/src/tq/tqScan.c @@ -74,14 +74,14 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs qTaskInfo_t task = pExec->task; if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) { - tqDebug("prepare scan failed, return, consumer:0x%"PRIx64, pHandle->consumerId); + tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId); if (pOffset->type == TMQ_OFFSET__LOG) { pRsp->rspOffset = *pOffset; return code; } else { tqOffsetResetToLog(pOffset, pHandle->snapshotVer); if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) { - tqDebug("prepare scan failed, return, consumer:0x%"PRIx64, pHandle->consumerId); + tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId); pRsp->rspOffset = *pOffset; return code; } @@ -92,7 +92,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs SSDataBlock* pDataBlock = NULL; uint64_t ts = 0; - tqDebug("vgId:%d, tmq task start to execute, consumer:0x%" PRIx64, vgId, pHandle->consumerId); + tqDebug("vgId:%d, tmq task start to execute, consumer:0x%"PRIx64, vgId, pHandle->consumerId); code = qExecTask(task, &pDataBlock, &ts); if (code != TSDB_CODE_SUCCESS) { diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index f99c7de93d..9b8f034e44 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -107,6 +107,7 @@ int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t uint64_t tableListGetSize(const STableListInfo* pTableList); uint64_t tableListGetSuid(const STableListInfo* pTableList); STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index); +int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex); size_t getResultRowSize(struct SqlFunctionCtx* pCtx, int32_t numOfOutput); void initResultRowInfo(SResultRowInfo* pResultRowInfo); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 3f519568c4..0cb9955056 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -143,10 +143,7 @@ typedef struct { SQueryTableDataCond tableCond; int64_t fillHistoryVer1; int64_t fillHistoryVer2; - - // int8_t triggerSaved; - // int64_t deleteMarkSaved; - SStreamState* pState; + SStreamState* pState; } SStreamTaskInfo; typedef struct { @@ -168,15 +165,14 @@ typedef struct STaskStopInfo { } STaskStopInfo; struct SExecTaskInfo { - STaskIdInfo id; - uint32_t status; - STimeWindow window; - STaskCostInfo cost; - int64_t owner; // if it is in execution - int32_t code; - int32_t qbufQuota; // total available buffer (in KB) during execution query - - int64_t version; // used for stream to record wal version, why not move to sschemainfo + STaskIdInfo id; + uint32_t status; + STimeWindow window; + STaskCostInfo cost; + int64_t owner; // if it is in execution + int32_t code; + int32_t qbufQuota; // total available buffer (in KB) during execution query + int64_t version; // used for stream to record wal version, why not move to sschemainfo SStreamTaskInfo streamInfo; SSchemaInfo schemaInfo; STableListInfo* pTableInfoList; // this is a table list @@ -188,6 +184,7 @@ struct SExecTaskInfo { SLocalFetch localFetch; SArray* pResultBlockList; // result block list STaskStopInfo stopInfo; + SRWLatch lock; // secure the access of STableListInfo }; enum { @@ -485,12 +482,6 @@ typedef struct SStreamScanInfo { } SStreamScanInfo; typedef struct { - // int8_t subType; - // bool withMeta; - // int64_t suid; - // int64_t snapVersion; - // void *metaInfo; - // void *dataInfo; SVnode* vnode; SSDataBlock pRes; // result SSDataBlock STsdbReader* dataReader; @@ -693,6 +684,8 @@ typedef struct SStreamFillOperatorInfo { #define OPTR_IS_OPENED(_optr) (((_optr)->status & OP_OPENED) == OP_OPENED) #define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED) +SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model, char* dbFName); + SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup, __optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain); int32_t optrDummyOpenFn(SOperatorInfo* pOperator); diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 0081ffcb0b..7d318786ba 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -34,7 +34,7 @@ struct STableListInfo { int32_t numOfOuputGroups; // the data block will be generated one by one int32_t* groupOffset; // keep the offset value for each group in the tableList SArray* pTableList; - SHashObj* map; // speedup acquire the tableQueryInfo by table uid + SHashObj* map; // speedup acquire the tableQueryInfo by table uid uint64_t suid; }; @@ -1804,6 +1804,21 @@ STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) return taosArrayGet(pTableList->pTableList, index); } +int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) { + int32_t numOfTables = taosArrayGetSize(pTableList->pTableList); + if (startIndex >= numOfTables) { + return -1; + } + + for (int32_t i = startIndex; i < numOfTables; ++i) { + STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i); + if (p->uid == uid) { + return i; + } + } + return -1; +} + uint64_t getTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) { int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid)); ASSERT(pTableList->map != NULL && slot != NULL); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 85f17c0d53..656ffda0ca 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -242,30 +242,28 @@ int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, return code; } -qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, SSchemaWrapper** pSchema) { - if (msg == NULL) { - // create raw scan - SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo)); +qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, uint64_t id) { + if (msg == NULL) { // create raw scan + SExecTaskInfo* pTaskInfo = doCreateExecTaskInfo(0, id, vgId, OPTR_EXEC_MODEL_QUEUE, ""); if (NULL == pTaskInfo) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); - - pTaskInfo->cost.created = taosGetTimestampUs(); - pTaskInfo->execModel = OPTR_EXEC_MODEL_QUEUE; pTaskInfo->pRoot = createRawScanOperatorInfo(pReaderHandle, pTaskInfo); if (NULL == pTaskInfo->pRoot) { terrno = TSDB_CODE_OUT_OF_MEMORY; taosMemoryFree(pTaskInfo); return NULL; } + + qDebug("create raw scan task info completed, vgId:%d, %s", vgId, GET_TASKID(pTaskInfo)); return pTaskInfo; } struct SSubplan* pPlan = NULL; - int32_t code = qStringToSubplan(msg, &pPlan); + + int32_t code = qStringToSubplan(msg, &pPlan); if (code != TSDB_CODE_SUCCESS) { terrno = code; return NULL; @@ -292,9 +290,6 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int3 } } - if (pSchema) { - *pSchema = tCloneSSchemaWrapper(((SExecTaskInfo*)pTaskInfo)->schemaInfo.qsw); - } return pTaskInfo; } @@ -410,6 +405,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo } STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList; + taosWLockLatch(&pTaskInfo->lock); for (int32_t i = 0; i < numOfQualifiedTables; ++i) { uint64_t* uid = taosArrayGet(qa, i); @@ -424,6 +420,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo if (code != TSDB_CODE_SUCCESS) { taosMemoryFree(keyBuf); taosArrayDestroy(qa); + taosWUnLockLatch(&pTaskInfo->lock); return code; } } @@ -445,6 +442,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo tableListAddTableInfo(pTableListInfo, keyInfo.uid, keyInfo.groupId); } + taosWUnLockLatch(&pTaskInfo->lock); if (keyBuf != NULL) { taosMemoryFree(keyBuf); } @@ -452,7 +450,9 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo taosArrayDestroy(qa); } else { // remove the table id in current list qDebug(" %d remove child tables from the stream scanner", (int32_t)taosArrayGetSize(tableIdList)); + taosWLockLatch(&pTaskInfo->lock); code = tqReaderRemoveTbUidList(pScanInfo->tqReader, tableIdList); + taosWUnLockLatch(&pTaskInfo->lock); } return code; @@ -1000,6 +1000,7 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) { } return 0; } + bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; return pTaskInfo->streamInfo.recoverScanFinished; @@ -1080,8 +1081,11 @@ int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit) { } int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) { - SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; - SOperatorInfo* pOperator = pTaskInfo->pRoot; + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; + SOperatorInfo* pOperator = pTaskInfo->pRoot; + STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList; + const char* id = GET_TASKID(pTaskInfo); + pTaskInfo->streamInfo.prepareStatus = *pOffset; pTaskInfo->streamInfo.returned = 0; @@ -1094,21 +1098,24 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT // TODO add more check if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { - if(pOperator->numOfDownstream != 1){ - qError("pOperator->numOfDownstream != 1:%d", pOperator->numOfDownstream); + if (pOperator->numOfDownstream != 1) { + qError("invalid operator, number of downstream:%d, %s", pOperator->numOfDownstream, id); return -1; } pOperator = pOperator->pDownstream[0]; } SStreamScanInfo* pInfo = pOperator->info; + STableScanInfo* pScanInfo = pInfo->pTableScanOp->info; + STableScanBase* pScanBaseInfo = &pScanInfo->base; + if (pOffset->type == TMQ_OFFSET__LOG) { - STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; - tsdbReaderClose(pTSInfo->base.dataReader); - pTSInfo->base.dataReader = NULL; + tsdbReaderClose(pScanBaseInfo->dataReader); + pScanBaseInfo->dataReader = NULL; + // let's seek to the next version in wal file if (tqSeekVer(pInfo->tqReader, pOffset->version + 1, pTaskInfo->id.str) < 0) { - qError("tqSeekVer failed ver:%" PRId64, pOffset->version + 1); + qError("tqSeekVer failed ver:%"PRId64", %s", pOffset->version + 1, id); return -1; } } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { @@ -1116,121 +1123,125 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT // those data are from the snapshot in tsdb, besides the data in the wal file. int64_t uid = pOffset->uid; int64_t ts = pOffset->ts; + int32_t index = 0; + + // this value may be changed if new tables are created + taosRLockLatch(&pTaskInfo->lock); + int32_t numOfTables = tableListGetSize(pTableListInfo); if (uid == 0) { - if (tableListGetSize(pTaskInfo->pTableInfoList) != 0) { - STableKeyInfo* pTableInfo = tableListGetInfo(pTaskInfo->pTableInfoList, 0); + if (numOfTables != 0) { + STableKeyInfo* pTableInfo = tableListGetInfo(pTableListInfo, 0); uid = pTableInfo->uid; ts = INT64_MIN; + pScanInfo->currentTable = 0; } else { - qError("uid == 0 and tablelist size is 0"); + taosRUnLockLatch(&pTaskInfo->lock); + qError("no table in table list, %s", id); return -1; } } - /*if (pTaskInfo->streamInfo.lastStatus.type != TMQ_OFFSET__SNAPSHOT_DATA ||*/ - /*pTaskInfo->streamInfo.lastStatus.uid != uid || pTaskInfo->streamInfo.lastStatus.ts != ts) {*/ - STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; - int32_t numOfTables = tableListGetSize(pTaskInfo->pTableInfoList); - - qDebug("switch to table uid:%" PRId64 " ts:%" PRId64 "% "PRId64 " rows returned", uid, ts, pInfo->pTableScanOp->resultInfo.totalRows); pInfo->pTableScanOp->resultInfo.totalRows = 0; - bool found = false; - for (int32_t i = 0; i < numOfTables; i++) { - STableKeyInfo* pTableInfo = tableListGetInfo(pTaskInfo->pTableInfoList, i); - if (pTableInfo->uid == uid) { - found = true; - pTableScanInfo->currentTable = i; - break; - } - } + // start from current accessed position + // we cannot start from the pScanInfo->currentTable, since the commit offset may cause the rollback of the start + // position, let's find it from the beginning. + index = tableListFind(pTableListInfo, uid, 0); + taosRUnLockLatch(&pTaskInfo->lock); - // TODO after dropping table, table may not found - if(!found){ - qError("uid not found in tablelist %" PRId64, uid); + if (index >= 0) { + pScanInfo->currentTable = index; + } else { + qError("vgId:%d uid:%" PRIu64 " not found in table list, total:%d, index:%d %s", pTaskInfo->id.vgId, uid, + numOfTables, pScanInfo->currentTable, id); return -1; } - if (pTableScanInfo->base.dataReader == NULL) { - STableKeyInfo* pList = tableListGetInfo(pTaskInfo->pTableInfoList, 0); - int32_t num = tableListGetSize(pTaskInfo->pTableInfoList); + STableKeyInfo keyInfo = {.uid = uid}; + int64_t oldSkey = pScanBaseInfo->cond.twindows.skey; - if (tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &pTableScanInfo->base.cond, pList, num, - pTableScanInfo->pResBlock, &pTableScanInfo->base.dataReader, NULL) < 0 || - pTableScanInfo->base.dataReader == NULL) { - qError("tsdbReaderOpen failed. uid:%" PRIi64, pOffset->uid); + // let's start from the next ts that returned to consumer. + pScanBaseInfo->cond.twindows.skey = ts + 1; + pScanInfo->scanTimes = 0; + + if (pScanBaseInfo->dataReader == NULL) { + int32_t code = tsdbReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond, &keyInfo, 1, + pScanInfo->pResBlock, &pScanBaseInfo->dataReader, id); + if (code != TSDB_CODE_SUCCESS) { + qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id); + terrno = code; return -1; } + + qDebug("tsdb reader created with offset(snapshot) uid:%" PRId64 " ts:%" PRId64 " table index:%d, total:%d, %s", + uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id); + } else { + tsdbSetTableList(pScanBaseInfo->dataReader, &keyInfo, 1); + tsdbReaderReset(pScanBaseInfo->dataReader, &pScanBaseInfo->cond); + qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 " table index:%d numOfTable:%d, %s", + uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id); } - STableKeyInfo tki = {.uid = uid}; - tsdbSetTableList(pTableScanInfo->base.dataReader, &tki, 1); - int64_t oldSkey = pTableScanInfo->base.cond.twindows.skey; - pTableScanInfo->base.cond.twindows.skey = ts + 1; - tsdbReaderReset(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond); - pTableScanInfo->base.cond.twindows.skey = oldSkey; - pTableScanInfo->scanTimes = 0; - - qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 ", table cur set to %d , all table num %d", uid, - ts, pTableScanInfo->currentTable, numOfTables); + // restore the key value + pScanBaseInfo->cond.twindows.skey = oldSkey; } else { - qError("invalid pOffset->type:%d", pOffset->type); - return -1; - } - } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { - SStreamRawScanInfo* pInfo = pOperator->info; - SSnapContext* sContext = pInfo->sContext; - if (setForSnapShot(sContext, pOffset->uid) != 0) { - qError("setDataForSnapShot error. uid:%" PRIi64, pOffset->uid); + qError("invalid pOffset->type:%d, %s", pOffset->type, id); return -1; } - SMetaTableInfo mtInfo = getUidfromSnapShot(sContext); - tsdbReaderClose(pInfo->dataReader); - pInfo->dataReader = NULL; + } else { // subType == TOPIC_SUB_TYPE__TABLE/TOPIC_SUB_TYPE__DB + if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { + SStreamRawScanInfo* pInfo = pOperator->info; + SSnapContext* sContext = pInfo->sContext; + if (setForSnapShot(sContext, pOffset->uid) != 0) { + qError("setDataForSnapShot error. uid:%" PRId64" , %s", pOffset->uid, id); + return -1; + } - cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); - tableListClear(pTaskInfo->pTableInfoList); + SMetaTableInfo mtInfo = getUidfromSnapShot(sContext); + tsdbReaderClose(pInfo->dataReader); + pInfo->dataReader = NULL; - if (mtInfo.uid == 0) { - return 0; // no data + cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); + tableListClear(pTableListInfo); + + if (mtInfo.uid == 0) { + return 0; // no data + } + + initQueryTableDataCondForTmq(&pTaskInfo->streamInfo.tableCond, sContext, &mtInfo); + pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts; + + tableListAddTableInfo(pTableListInfo, mtInfo.uid, 0); + + STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0); + int32_t size = tableListGetSize(pTableListInfo); + + tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, &pInfo->dataReader, NULL); + + cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); + strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName); + tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema); + pTaskInfo->streamInfo.schema = mtInfo.schema; + + qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64" %s", mtInfo.uid, pOffset->ts, id); + } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) { + SStreamRawScanInfo* pInfo = pOperator->info; + SSnapContext* sContext = pInfo->sContext; + if (setForSnapShot(sContext, pOffset->uid) != 0) { + qError("setForSnapShot error. uid:%" PRIu64 " ,version:%" PRId64, pOffset->uid, pOffset->version); + return -1; + } + qDebug("tmqsnap qStreamPrepareScan snapshot meta uid:%" PRId64 " ts %" PRId64 " %s", pOffset->uid, pOffset->ts, id); + } else if (pOffset->type == TMQ_OFFSET__LOG) { + SStreamRawScanInfo* pInfo = pOperator->info; + tsdbReaderClose(pInfo->dataReader); + pInfo->dataReader = NULL; + qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id); } - - initQueryTableDataCondForTmq(&pTaskInfo->streamInfo.tableCond, sContext, &mtInfo); - pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts; - - if (pTaskInfo->pTableInfoList == NULL) { - pTaskInfo->pTableInfoList = tableListCreate(); - } - - tableListAddTableInfo(pTaskInfo->pTableInfoList, mtInfo.uid, 0); - - STableKeyInfo* pList = tableListGetInfo(pTaskInfo->pTableInfoList, 0); - int32_t size = tableListGetSize(pTaskInfo->pTableInfoList); - - tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, &pInfo->dataReader, NULL); - - cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); - strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName); - tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema); - pTaskInfo->streamInfo.schema = mtInfo.schema; - - qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64, mtInfo.uid, pOffset->ts); - } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) { - SStreamRawScanInfo* pInfo = pOperator->info; - SSnapContext* sContext = pInfo->sContext; - if (setForSnapShot(sContext, pOffset->uid) != 0) { - qError("setForSnapShot error. uid:%" PRIu64 " ,version:%" PRId64, pOffset->uid, pOffset->version); - return -1; - } - qDebug("tmqsnap qStreamPrepareScan snapshot meta uid:%" PRId64 " ts %" PRId64, pOffset->uid, pOffset->ts); - } else if (pOffset->type == TMQ_OFFSET__LOG) { - SStreamRawScanInfo* pInfo = pOperator->info; - tsdbReaderClose(pInfo->dataReader); - pInfo->dataReader = NULL; - qDebug("tmqsnap qStreamPrepareScan snapshot log"); } + return 0; } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 24a26d575a..c2a068e1cb 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1974,7 +1974,7 @@ char* buildTaskId(uint64_t taskId, uint64_t queryId) { return p; } -static SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model, char* dbFName) { +SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, int32_t vgId, EOPTR_EXEC_MODEL model, char* dbFName) { SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo)); if (pTaskInfo == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -1982,6 +1982,7 @@ static SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, in } setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); + pTaskInfo->cost.created = taosGetTimestampUs(); pTaskInfo->schemaInfo.dbname = taosStrdup(dbFName); pTaskInfo->execModel = model; @@ -1989,6 +1990,7 @@ static SExecTaskInfo* doCreateExecTaskInfo(uint64_t queryId, uint64_t taskId, in pTaskInfo->stopInfo.pStopInfo = taosArrayInit(4, sizeof(SExchangeOpStopInfo)); pTaskInfo->pResultBlockList = taosArrayInit(128, POINTER_BYTES); + taosInitRWLatch(&pTaskInfo->lock); pTaskInfo->id.vgId = vgId; pTaskInfo->id.queryId = queryId; pTaskInfo->id.str = buildTaskId(taskId, queryId); @@ -2464,7 +2466,6 @@ int32_t createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHand goto _complete; } - (*pTaskInfo)->cost.created = taosGetTimestampUs(); return TSDB_CODE_SUCCESS; _complete: diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 870c96811b..84317c825b 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -779,13 +779,14 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { // if no data, switch to next table and continue scan pInfo->currentTable++; if (pInfo->currentTable >= numOfTables) { + qDebug("all table checked in table list, total:%d, return NULL, %s", numOfTables, GET_TASKID(pTaskInfo)); return NULL; } STableKeyInfo* pTableInfo = tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->currentTable); tsdbSetTableList(pInfo->base.dataReader, pTableInfo, 1); - qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d %s", pTableInfo->uid, numOfTables, - pInfo->currentTable, pTaskInfo->id.str); + qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", pTableInfo->uid, numOfTables, + pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo)); tsdbReaderReset(pInfo->base.dataReader, &pInfo->base.cond); pInfo->scanTimes = 0; @@ -1595,19 +1596,16 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SStreamScanInfo* pInfo = pOperator->info; + const char* id = GET_TASKID(pTaskInfo); - qDebug("start to exec queue scan"); + qDebug("start to exec queue scan, %s", id); if (pTaskInfo->streamInfo.submit.msgStr != NULL) { - if (pInfo->tqReader->msg2.msgStr == NULL) { - /*pInfo->tqReader->pMsg = pTaskInfo->streamInfo.pReq;*/ - /*const SSubmitReq* pSubmit = pInfo->tqReader->pMsg;*/ - /*if (tqReaderSetDataMsg(pInfo->tqReader, pSubmit, 0) < 0) {*/ - /*void* msgStr = pTaskInfo->streamInfo.*/ + if (pInfo->tqReader->msg2.msgStr == NULL) { SPackedData submit = pTaskInfo->streamInfo.submit; if (tqReaderSetSubmitReq2(pInfo->tqReader, submit.msgStr, submit.msgLen, submit.ver) < 0) { - qError("submit msg messed up when initing stream submit block %p", submit.msgStr); + qError("submit msg messed up when initing stream submit block %p, %s", submit.msgStr, id); pInfo->tqReader->msg2 = (SPackedData){0}; pInfo->tqReader->setMsg = 0; ASSERT(0); @@ -1641,18 +1639,20 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) { SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp); if (pResult && pResult->info.rows > 0) { - qDebug("queue scan tsdb return %d rows min:%" PRId64 " max:%" PRId64 " wal curVersion:%" PRId64, pResult->info.rows, - pResult->info.window.skey, pResult->info.window.ekey, pInfo->tqReader->pWalReader->curVersion); + qDebug("queue scan tsdb return %d rows min:%" PRId64 " max:%" PRId64 " wal curVersion:%" PRId64" %s", pResult->info.rows, + pResult->info.window.skey, pResult->info.window.ekey, pInfo->tqReader->pWalReader->curVersion, id); pTaskInfo->streamInfo.returned = 1; return pResult; } else { + // no data has return already, try to extract data in the WAL if (!pTaskInfo->streamInfo.returned) { STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; tsdbReaderClose(pTSInfo->base.dataReader); - qDebug("3"); + pTSInfo->base.dataReader = NULL; tqOffsetResetToLog(&pTaskInfo->streamInfo.prepareStatus, pTaskInfo->streamInfo.snapshotVer); - qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1); + + qDebug("queue scan tsdb over, switch to wal ver:%" PRId64 " %s", pTaskInfo->streamInfo.snapshotVer + 1, id); if (tqSeekVer(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1, pTaskInfo->id.str) < 0) { tqOffsetResetToLog(&pTaskInfo->streamInfo.lastStatus, pTaskInfo->streamInfo.snapshotVer); return NULL; @@ -1669,7 +1669,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { if (tqNextBlock(pInfo->tqReader, &ret) < 0) { // if the end is reached, terrno is 0 if (terrno != 0) { - qError("failed to get next log block since %s", terrstr()); + qError("failed to get next log block since %s, %s", terrstr(), id); } } @@ -1684,9 +1684,6 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { } else if (ret.fetchType == FETCH_TYPE__META) { qError("unexpected ret.fetchType:%d", ret.fetchType); continue; - // pTaskInfo->streamInfo.lastStatus = ret.offset; - // pTaskInfo->streamInfo.metaBlk = ret.meta; - // return NULL; } else if (ret.fetchType == FETCH_TYPE__NONE || (ret.fetchType == FETCH_TYPE__SEP && pOperator->status == OP_EXEC_RECV)) { pTaskInfo->streamInfo.lastStatus = ret.offset; @@ -1698,7 +1695,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { } } } else { - qError("unexpected streamInfo prepare type: %d", pTaskInfo->streamInfo.prepareStatus.type); + qError("unexpected streamInfo prepare type: %d %s", pTaskInfo->streamInfo.prepareStatus.type, id); return NULL; } } diff --git a/tests/system-test/2-query/columnLenUpdated.py b/tests/system-test/2-query/columnLenUpdated.py index 3b87cbe22a..e43b32a716 100644 --- a/tests/system-test/2-query/columnLenUpdated.py +++ b/tests/system-test/2-query/columnLenUpdated.py @@ -83,7 +83,7 @@ class TDTestCase: def init(self, conn, logSql, replicaVar=1): self.replicaVar = int(replicaVar) tdLog.debug(f"start to excute {__file__}") - tdSql.init(conn.cursor()) + tdSql.init(conn.cursor(), True) def getBuildPath(self): selfPath = os.path.dirname(os.path.realpath(__file__)) diff --git a/tests/system-test/7-tmq/subscribeDb.py b/tests/system-test/7-tmq/subscribeDb.py index 0fa9bcfbd4..41db943f83 100644 --- a/tests/system-test/7-tmq/subscribeDb.py +++ b/tests/system-test/7-tmq/subscribeDb.py @@ -13,11 +13,11 @@ from util.dnodes import * class TDTestCase: hostname = socket.gethostname() - #rpcDebugFlagVal = '143' + # rpcDebugFlagVal = '143' #clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''} #clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal #updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''} - #updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal + # updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal #print ("===================: ", updatecfgDict) def init(self, conn, logSql, replicaVar=1): diff --git a/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py b/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py index 87832ac0ef..11fc7dbcc0 100644 --- a/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py +++ b/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py @@ -16,6 +16,8 @@ sys.path.append("./7-tmq") from tmqCommon import * class TDTestCase: + updatecfgDict = {"tsdbDebugFlag":135} + def __init__(self): self.vgroups = 4 self.ctbNum = 10