diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 9c84d0dd99e9823de47d3420c610770bcf8c9197..a14f7eff8a0f4b8f455a96a3a83a6be09c4995f0 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -47,6 +47,7 @@ typedef enum EStreamType { STREAM_GET_ALL, STREAM_DELETE, STREAM_RETRIEVE, + STREAM_PUSH_DATA, } EStreamType; typedef struct { @@ -71,7 +72,7 @@ typedef struct SColumnDataAgg { typedef struct SDataBlockInfo { STimeWindow window; - int32_t rows; // todo hide this attribute + int32_t rows; // todo hide this attribute int32_t rowSize; uint64_t uid; // the uid of table, from which current data block comes uint16_t blockId; // block id, generated by physical planner diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index a47810e7c1a918ef4ce9795fdbd08bf45ce363ce..594344ba8a508e4e35260e122e9ab9ac1f8e25c3 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -58,6 +58,7 @@ enum { enum { STREAM_INPUT__DATA_SUBMIT = 1, STREAM_INPUT__DATA_BLOCK, + STREAM_INPUT__DATA_RETRIEVE, STREAM_INPUT__TRIGGER, STREAM_INPUT__CHECKPOINT, STREAM_INPUT__DROP, diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index 9d92865e4780fc5f0f9ea78ab9fe7356e0c0c661..56d063ae5141ac203d905a8d58c2d588f018e683 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -111,7 +111,7 @@ int32_t streamTaskEnqueue(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* // enqueue if (pData != NULL) { - pData->type = STREAM_DATA_TYPE_SSDATA_BLOCK; + pData->type = STREAM_INPUT__DATA_BLOCK; pData->srcVgId = pReq->dataSrcVgId; // decode /*pData->blocks = pReq->data;*/ @@ -146,7 +146,7 @@ int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq, // enqueue if (pData != NULL) { - pData->type = STREAM_DATA_TYPE_SSDATA_BLOCK; + pData->type = STREAM_INPUT__DATA_RETRIEVE; pData->srcVgId = 0; // decode /*pData->blocks = pReq->data;*/ @@ -170,7 +170,7 @@ int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq, pCont->rspToTaskId = pReq->srcTaskId; pCont->rspFromTaskId = pReq->dstTaskId; pRsp->pCont = buf; - pRsp->contLen = sizeof(SMsgHead) + sizeof(SStreamDispatchRsp); + pRsp->contLen = sizeof(SMsgHead) + sizeof(SStreamRetrieveRsp); tmsgSendRsp(pRsp); return status == TASK_INPUT_STATUS__NORMAL ? 0 : -1; } diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 1f51a927e7a7d091c1018e104f49d31f57dde4cf..0c9d46f0555009686275b036ec702a641cc7ad4f 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -300,7 +300,7 @@ int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) { atomic_store_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL); return 0; } - ASSERT(pBlock->type == STREAM_DATA_TYPE_SSDATA_BLOCK); + ASSERT(pBlock->type == STREAM_INPUT__DATA_BLOCK); qInfo("stream continue dispatching: task %d", pTask->taskId); diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index cb7dd241c1912d0f5e2dca88be9562e3fca095e3..8e7cac03a20208fac05848e9d2e8d63a860ca590 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -17,6 +17,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) { void* exec = pTask->exec.executor; + bool hasData = false; // set input SStreamQueueItem* pItem = (SStreamQueueItem*)data; @@ -27,7 +28,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) ASSERT(pTask->isDataScan); SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data; qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK, false); - } else if (pItem->type == STREAM_INPUT__DATA_BLOCK) { + } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) { SStreamDataBlock* pBlock = (SStreamDataBlock*)data; SArray* blocks = pBlock->blocks; qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK, false); @@ -43,7 +44,16 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) if (qExecTask(exec, &output, &ts) < 0) { ASSERT(false); } - if (output == NULL) break; + if (output == NULL) { + if (pItem->type == STREAM_INPUT__DATA_RETRIEVE && !hasData) { + SSDataBlock block = {0}; + block.info.type = STREAM_PUSH_DATA; + block.info.childId = pTask->selfChildId; + taosArrayPush(pRes, &block); + } + break; + } + hasData = true; if (output->info.type == STREAM_RETRIEVE) { if (streamBroadcastToChildren(pTask, output) < 0) { diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 1d169a08913d7afc87c1155b1adcc50b32f154c2..d1295667afb0ce9bfc36dfab385d5fc9a8b3627d 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -42,6 +42,9 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) { char fnameStr[WAL_FILE_LEN]; walBuildLogName(pWal, pFileInfo->firstVer, fnameStr); taosRemoveFile(fnameStr); + + walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr); + taosRemoveFile(fnameStr); } } walRemoveMeta(pWal); @@ -105,7 +108,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) { } walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr); - TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ); + TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); if (pIdxTFile == NULL) { taosThreadMutexUnlock(&pWal->mutex); @@ -126,7 +129,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) { ASSERT(entry.ver == ver); walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr); - TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ); + TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); if (pLogTFile == NULL) { // TODO taosThreadMutexUnlock(&pWal->mutex); @@ -307,8 +310,8 @@ int walRoll(SWal *pWal) { static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) { SWalIdxEntry entry = {.ver = ver, .offset = offset}; - /*int64_t idxOffset = taosLSeekFile(pWal->pWriteIdxTFile, 0, SEEK_CUR);*/ - /*wDebug("write index: ver: %ld, offset: %ld, at %ld", ver, offset, idxOffset);*/ + int64_t idxOffset = taosLSeekFile(pWal->pWriteIdxTFile, 0, SEEK_END); + wDebug("write index: ver: %ld, offset: %ld, at %ld", ver, offset, idxOffset); int64_t size = taosWriteFile(pWal->pWriteIdxTFile, &entry, sizeof(SWalIdxEntry)); if (size != sizeof(SWalIdxEntry)) { terrno = TAOS_SYSTEM_ERROR(errno);