diff --git a/example/src/tmq.c b/example/src/tmq.c index 8757104ad9..efb4d1830e 100644 --- a/example/src/tmq.c +++ b/example/src/tmq.c @@ -28,7 +28,7 @@ int32_t init_env() { return -1; } - TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1"); + TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2"); if (taos_errno(pRes) != 0) { printf("error in create db, reason:%s\n", taos_errstr(pRes)); return -1; @@ -42,25 +42,33 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)"); + pRes = + taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)"); if (taos_errno(pRes) != 0) { printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)"); + pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000)"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)"); + pRes = taos_query(pConn, "create table if not exists ct1 using st1 tags(2000)"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes)); return -1; } + + pRes = taos_query(pConn, "create table if not exists ct3 using st1 tags(3000)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); return 0; } @@ -82,12 +90,40 @@ int32_t create_topic() { /*const char* sql = "select * from tu1";*/ /*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));*/ - pRes = taos_query(pConn, "create topic test_stb_topic_1 as select * from tu1"); + pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1 from ct1"); if (taos_errno(pRes) != 0) { - printf("failed to create topic test_stb_topic_1, reason:%s\n", taos_errstr(pRes)); + printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); + +#if 0 + pRes = taos_query(pConn, "insert into tu1 values(now, 1, 1.0, 'bi1')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "insert into tu1 values(now+1d, 1, 1.0, 'bi1')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "insert into tu2 values(now, 2, 2.0, 'bi2')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "insert into tu2 values(now+1d, 2, 2.0, 'bi2')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); +#endif + taos_close(pConn); return 0; } @@ -115,7 +151,7 @@ tmq_t* build_consumer() { tmq_list_t* build_topic_list() { tmq_list_t* topic_list = tmq_list_new(); - tmq_list_append(topic_list, "test_stb_topic_1"); + tmq_list_append(topic_list, "topic_ctb_column"); return topic_list; } @@ -215,8 +251,8 @@ int main(int argc, char* argv[]) { if (argc > 1) { printf("env init\n"); code = init_env(); + create_topic(); } - create_topic(); tmq_t* tmq = build_consumer(); tmq_list_t* topic_list = build_topic_list(); /*perf_loop(tmq, topic_list);*/ diff --git a/example/src/tstream.c b/example/src/tstream.c index 40d8ff9b0b..8ffa932bd2 100644 --- a/example/src/tstream.c +++ b/example/src/tstream.c @@ -20,7 +20,7 @@ #include "taos.h" int32_t init_env() { - TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 7010); + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); if (pConn == NULL) { return -1; } @@ -65,7 +65,7 @@ int32_t init_env() { int32_t create_stream() { printf("create stream\n"); TAOS_RES* pRes; - TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 7010); + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); if (pConn == NULL) { return -1; } diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 51eabb7d61..b57de3660b 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -37,6 +37,14 @@ enum { TMQ_MSG_TYPE__EP_RSP, }; +enum { + STREAM_TRIGGER__AT_ONCE = 1, + STREAM_TRIGGER__WINDOW_CLOSE, + STREAM_TRIGGER__BY_COUNT, + STREAM_TRIGGER__BY_BATCH_COUNT, + STREAM_TRIGGER__BY_EVENT_TIME, +}; + typedef struct { uint32_t numOfTables; SArray* pGroupList; @@ -54,13 +62,16 @@ typedef struct SColumnDataAgg { } SColumnDataAgg; typedef struct SDataBlockInfo { - STimeWindow window; - int32_t rows; - int32_t rowSize; - int16_t numOfCols; - int16_t hasVarCol; - union {int64_t uid; int64_t blockId;}; - int64_t groupId; // no need to serialize + STimeWindow window; + int32_t rows; + int32_t rowSize; + int16_t numOfCols; + int16_t hasVarCol; + union { + int64_t uid; + int64_t blockId; + }; + int64_t groupId; // no need to serialize } SDataBlockInfo; typedef struct SSDataBlock { @@ -93,7 +104,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock); int32_t tEncodeDataBlocks(void** buf, const SArray* blocks); void* tDecodeDataBlocks(const void* buf, SArray** blocks); -void colDataDestroy(SColumnInfoData* pColData) ; +void colDataDestroy(SColumnInfoData* pColData); static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) { // WARNING: do not use info.numOfCols, @@ -239,7 +250,7 @@ typedef struct SSessionWindow { SColumn col; } SSessionWindow; -#define QUERY_ASC_FORWARD_STEP 1 +#define QUERY_ASC_FORWARD_STEP 1 #define QUERY_DESC_FORWARD_STEP -1 #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP) diff --git a/include/common/trow.h b/include/common/trow.h index df28bc9962..fb6d3ece44 100644 --- a/include/common/trow.h +++ b/include/common/trow.h @@ -47,21 +47,21 @@ extern "C" { #define TD_VTYPE_NONE 0x0U // none or unknown/undefined #define TD_VTYPE_NULL 0x01U // null val #define TD_VTYPE_NORM 0x02U // normal val: not none, not null -#define TD_VTYPE_MAX 0x03U // +#define TD_VTYPE_MAX 0x03U // #define TD_VTYPE_NONE_BYTE 0x0U #define TD_VTYPE_NULL_BYTE 0x55U #define TD_VTYPE_NORM_BYTE 0xAAU -#define TD_ROWS_ALL_NORM 0x01U +#define TD_ROWS_ALL_NORM 0x01U #define TD_ROWS_NULL_NORM 0x0U -#define TD_COL_ROWS_NORM(c) ((c)->bitmap == TD_ROWS_ALL_NORM) // all rows of SDataCol/SBlockCol is NORM +#define TD_COL_ROWS_NORM(c) ((c)->bitmap == TD_ROWS_ALL_NORM) // all rows of SDataCol/SBlockCol is NORM #define TD_SET_COL_ROWS_BTIMAP(c, v) ((c)->bitmap = (v)) -#define TD_SET_COL_ROWS_NORM(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM) -#define TD_SET_COL_ROWS_MISC(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM) +#define TD_SET_COL_ROWS_NORM(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM) +#define TD_SET_COL_ROWS_MISC(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM) -#define KvConvertRatio (0.9f) +#define KvConvertRatio (0.9f) #define isSelectKVRow(klen, tlen) ((klen) < ((tlen)*KvConvertRatio)) #ifdef TD_SUPPORT_BITMAP @@ -98,7 +98,7 @@ typedef void *SRow; typedef struct { TDRowValT valType; - void * val; + void *val; } SCellVal; typedef struct { @@ -158,43 +158,43 @@ typedef struct { int16_t nBitmaps; int16_t nBoundBitmaps; int32_t offset; - void * pBitmap; - void * pOffset; + void *pBitmap; + void *pOffset; int32_t extendedRowSize; } SRowBuilder; -#define TD_ROW_HEAD_LEN (sizeof(STSRow)) +#define TD_ROW_HEAD_LEN (sizeof(STSRow)) #define TD_ROW_NCOLS_LEN (sizeof(col_id_t)) -#define TD_ROW_INFO(r) ((r)->info) -#define TD_ROW_TYPE(r) ((r)->type) -#define TD_ROW_DELETE(r) ((r)->del) -#define TD_ROW_ENDIAN(r) ((r)->endian) -#define TD_ROW_SVER(r) ((r)->sver) -#define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow -#define TD_ROW_DATA(r) ((r)->data) -#define TD_ROW_LEN(r) ((r)->len) -#define TD_ROW_KEY(r) ((r)->ts) +#define TD_ROW_INFO(r) ((r)->info) +#define TD_ROW_TYPE(r) ((r)->type) +#define TD_ROW_DELETE(r) ((r)->del) +#define TD_ROW_ENDIAN(r) ((r)->endian) +#define TD_ROW_SVER(r) ((r)->sver) +#define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow +#define TD_ROW_DATA(r) ((r)->data) +#define TD_ROW_LEN(r) ((r)->len) +#define TD_ROW_KEY(r) ((r)->ts) #define TD_ROW_KEY_ADDR(r) (r) // N.B. If without STSchema, getExtendedRowSize() is used to get the rowMaxBytes and // (int32_t)ceil((double)nCols/TD_VTYPE_PARTS) should be added if TD_SUPPORT_BITMAP defined. #define TD_ROW_MAX_BYTES_FROM_SCHEMA(s) (schemaTLen(s) + TD_ROW_HEAD_LEN) -#define TD_ROW_SET_INFO(r, i) (TD_ROW_INFO(r) = (i)) -#define TD_ROW_SET_TYPE(r, t) (TD_ROW_TYPE(r) = (t)) -#define TD_ROW_SET_DELETE(r) (TD_ROW_DELETE(r) = 1) -#define TD_ROW_SET_SVER(r, v) (TD_ROW_SVER(r) = (v)) -#define TD_ROW_SET_LEN(r, l) (TD_ROW_LEN(r) = (l)) +#define TD_ROW_SET_INFO(r, i) (TD_ROW_INFO(r) = (i)) +#define TD_ROW_SET_TYPE(r, t) (TD_ROW_TYPE(r) = (t)) +#define TD_ROW_SET_DELETE(r) (TD_ROW_DELETE(r) = 1) +#define TD_ROW_SET_SVER(r, v) (TD_ROW_SVER(r) = (v)) +#define TD_ROW_SET_LEN(r, l) (TD_ROW_LEN(r) = (l)) #define TD_ROW_SET_NCOLS(r, n) (*(col_id_t *)TD_ROW_NCOLS(r) = (n)) #define TD_ROW_IS_DELETED(r) (TD_ROW_DELETE(r) == 1) -#define TD_IS_TP_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_TP) -#define TD_IS_KV_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_KV) -#define TD_IS_TP_ROW_T(t) ((t) == TD_ROW_TP) -#define TD_IS_KV_ROW_T(t) ((t) == TD_ROW_KV) +#define TD_IS_TP_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_TP) +#define TD_IS_KV_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_KV) +#define TD_IS_TP_ROW_T(t) ((t) == TD_ROW_TP) +#define TD_IS_KV_ROW_T(t) ((t) == TD_ROW_KV) -#define TD_BOOL_STR(b) ((b) ? "true" : "false") +#define TD_BOOL_STR(b) ((b) ? "true" : "false") #define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert)) #define TD_ROW_COL_IDX(r) POINTER_SHIFT(TD_ROW_DATA(r), sizeof(col_id_t)) @@ -275,7 +275,7 @@ static FORCE_INLINE int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TD } int16_t nBytes = colIdx / TD_VTYPE_PARTS; int16_t nOffset = colIdx & TD_VTYPE_OPTR; - char * pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes); + char *pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes); switch (nOffset) { case 0: *pDestByte = ((*pDestByte) & 0x3F) | (valType << 6); @@ -313,7 +313,7 @@ static FORCE_INLINE int32_t tdGetBitmapValType(void *pBitmap, int16_t colIdx, TD } int16_t nBytes = colIdx / TD_VTYPE_PARTS; int16_t nOffset = colIdx & TD_VTYPE_OPTR; - char * pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes); + char *pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes); switch (nOffset) { case 0: *pValType = (((*pDestByte) & 0xC0) >> 6); @@ -620,7 +620,7 @@ static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowVa if (tdValIsNorm(valType, val, colType)) { // ts key stored in STSRow.ts SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset); - char * ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row)); + char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row)); pColIdx->colId = colId; pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN @@ -638,7 +638,7 @@ static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowVa // NULL/None value else { SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset); - char * ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row)); + char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row)); pColIdx->colId = colId; pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN const void *nullVal = getNullValue(colType); @@ -775,8 +775,8 @@ static FORCE_INLINE int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, v typedef struct { STSchema *pSchema; - STSRow * pRow; - void * pBitmap; + STSRow *pRow; + void *pBitmap; uint32_t offset; col_id_t maxColId; col_id_t colIdx; // [PRIMARYKEY_TIMESTAMP_COL_ID, nSchemaCols], PRIMARYKEY_TIMESTAMP_COL_ID equals 1 @@ -881,7 +881,7 @@ static FORCE_INLINE bool tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colTy // internal static FORCE_INLINE bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType, col_id_t *nIdx, SCellVal *pVal) { - STSRow * pRow = pIter->pRow; + STSRow *pRow = pIter->pRow; SKvRowIdx *pKvIdx = NULL; bool colFound = false; col_id_t kvNCols = tdRowGetNCols(pRow); @@ -939,7 +939,6 @@ static FORCE_INLINE bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col while (pIter->colIdx <= pSchema->numOfCols) { pCol = &pSchema->columns[pIter->colIdx]; if (colId == pCol->colId) { - ++pIter->colIdx; break; } else if (colId < pCol->colId) { ++pIter->colIdx; @@ -948,7 +947,8 @@ static FORCE_INLINE bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col return false; } } - return tdGetTpRowDataOfCol(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal); + tdGetTpRowDataOfCol(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal); + ++pIter->colIdx; } else if (TD_IS_KV_ROW(pIter->pRow)) { return tdGetKvRowValOfColEx(pIter, colId, colType, &pIter->kvIdx, pVal); } else { @@ -1076,7 +1076,7 @@ typedef struct { typedef struct { STSchema *pSchema; - STSRow * pRow; + STSRow *pRow; } STSRowReader; typedef struct { diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 8b1ac3ed8d..e635227eaa 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -61,16 +61,16 @@ extern "C" { } \ } -#define WAL_HEAD_VER 0 +#define WAL_HEAD_VER 0 #define WAL_NOSUFFIX_LEN 20 -#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1) -#define WAL_LOG_SUFFIX "log" +#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1) +#define WAL_LOG_SUFFIX "log" #define WAL_INDEX_SUFFIX "idx" -#define WAL_REFRESH_MS 1000 -#define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead)) -#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) -#define WAL_FILE_LEN (WAL_PATH_LEN + 32) -#define WAL_MAGIC 0xFAFBFCFDULL +#define WAL_REFRESH_MS 1000 +#define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead)) +#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) +#define WAL_FILE_LEN (WAL_PATH_LEN + 32) +#define WAL_MAGIC 0xFAFBFCFDULL #define WAL_CUR_FAILED 1 @@ -150,14 +150,15 @@ typedef struct SWal { } SWal; // WAL HANDLE typedef struct SWalReadHandle { - SWal *pWal; - TdFilePtr pReadLogTFile; - TdFilePtr pReadIdxTFile; - int64_t curFileFirstVer; - int64_t curVersion; - int64_t capacity; - int64_t status; // if cursor valid - SWalHead *pHead; + SWal *pWal; + TdFilePtr pReadLogTFile; + TdFilePtr pReadIdxTFile; + int64_t curFileFirstVer; + int64_t curVersion; + int64_t capacity; + int64_t status; // if cursor valid + TdThreadMutex mutex; + SWalHead *pHead; } SWalReadHandle; #pragma pack(pop) @@ -191,6 +192,7 @@ int32_t walEndSnapshot(SWal *); SWalReadHandle *walOpenReadHandle(SWal *); void walCloseReadHandle(SWalReadHandle *); int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver); +int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead); // deprecated #if 0 diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index 1fa267ae7e..5d1e95be7c 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -255,6 +255,7 @@ void tmqClearUnhandleMsg(tmq_t* tmq) { break; } + msg = NULL; taosReadAllQitems(tmq->mqueue, tmq->qall); while (1) { taosGetQitem(tmq->qall, (void**)&msg); @@ -787,7 +788,7 @@ void tmqShowMsg(tmq_message_t* tmq_message) { static bool noPrintSchema; char pBuf[128]; SMqPollRsp* pRsp = &tmq_message->msg; - int32_t colNum = pRsp->schema->nCols; + int32_t colNum = 2; if (!noPrintSchema) { printf("|"); for (int32_t i = 0; i < colNum; i++) { @@ -838,6 +839,7 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) { int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch; int32_t tmqEpoch = atomic_load_32(&tmq->epoch); if (msgEpoch < tmqEpoch) { + /*printf("discard rsp epoch %d, current epoch %d\n", msgEpoch, tmqEpoch);*/ tsem_post(&tmq->rspSem); tscWarn("discard rsp epoch %d, current epoch %d", msgEpoch, tmqEpoch); return 0; @@ -886,6 +888,9 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) { goto WRITE_QUEUE_FAIL; } + tscError("tmq recv poll: vg %d, req offset %ld, rsp offset %ld", pParam->pVg->vgId, pRsp->msg.reqOffset, + pRsp->msg.rspOffset); + pRsp->vg = pParam->pVg; taosWriteQitem(tmq->mqueue, pRsp); atomic_add_fetch_32(&tmq->readyRequest, 1); @@ -902,6 +907,7 @@ WRITE_QUEUE_FAIL: bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) { /*printf("call update ep %d\n", epoch);*/ + /*printf("tmq update ep epoch %d to epoch %d\n", tmq->epoch, epoch);*/ bool set = false; int32_t topicNumGet = taosArrayGetSize(pRsp->topics); char vgKey[TSDB_TOPIC_FNAME_LEN + 22]; @@ -932,6 +938,7 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) { for (int32_t k = 0; k < vgNumCur; k++) { SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, k); sprintf(vgKey, "%s:%d", topic.topicName, pVgCur->vgId); + /*printf("epoch %d vg %d build %s\n", epoch, pVgCur->vgId, vgKey);*/ taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t)); } break; @@ -945,9 +952,12 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) { sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId); int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey)); int64_t offset = pVgEp->offset; + /*printf("epoch %d vg %d offset og to %ld\n", epoch, pVgEp->vgId, offset);*/ if (pOffset != NULL) { offset = *pOffset; + /*printf("epoch %d vg %d found %s\n", epoch, pVgEp->vgId, vgKey);*/ } + /*printf("epoch %d vg %d offset set to %ld\n", epoch, pVgEp->vgId, offset);*/ SMqClientVg clientVg = { .pollCnt = 0, .currentOffset = offset, @@ -1195,6 +1205,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) { SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j); int32_t vgStatus = atomic_val_compare_exchange_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE, TMQ_VG_STATUS__WAIT); if (vgStatus != TMQ_VG_STATUS__IDLE) { + /*printf("skip vg %d\n", pVg->vgId);*/ continue; } SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, blockingTime, pTopic, pVg); @@ -1238,6 +1249,8 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) { int64_t transporterId = 0; /*printf("send poll\n");*/ atomic_add_fetch_32(&tmq->waitingRequest, 1); + /*tscDebug("tmq send poll: vg %d, req offset %ld", pVg->vgId, pVg->currentOffset);*/ + /*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/ asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo); pVg->pollCnt++; tmq->pollCnt++; diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 6976d83abd..7891c690e3 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -735,6 +735,9 @@ typedef struct { int8_t createdBy; // STREAM_CREATED_BY__USER or SMA int32_t fixedSinkVgId; // 0 for shuffle int64_t smaId; // 0 for unused + int8_t trigger; + int32_t triggerParam; + int64_t waterMark; char* sql; char* logicalPlan; char* physicalPlan; diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index e5d1f952a8..aa4a9fc1de 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -194,7 +194,7 @@ void tqClose(STQ*); int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t version); int tqCommit(STQ*); -int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg); +int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId); int32_t tqProcessSetConnReq(STQ* pTq, char* msg); int32_t tqProcessRebReq(STQ* pTq, char* msg); int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId); diff --git a/source/dnode/vnode/src/meta/metaTbUid.c b/source/dnode/vnode/src/meta/metaTbUid.c index cad1eba134..b488630024 100644 --- a/source/dnode/vnode/src/meta/metaTbUid.c +++ b/source/dnode/vnode/src/meta/metaTbUid.c @@ -27,5 +27,5 @@ void metaCloseUidGnrt(SMeta *pMeta) { /* TODO */ tb_uid_t metaGenerateUid(SMeta *pMeta) { // Generate a new table UID - return ++(pMeta->uidGnrt.nextUid); + return tGenIdPI64(); } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 06a2350aab..f34fa9e19c 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -250,7 +250,7 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu return 0; } -int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { +int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { SMqPollReq* pReq = pMsg->pCont; int64_t consumerId = pReq->consumerId; int64_t fetchOffset; @@ -264,6 +264,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { fetchOffset = pReq->currentOffset + 1; } + printf("tmq poll vg %d req %ld %ld\n", pTq->pVnode->vgId, pReq->currentOffset, fetchOffset); + SMqPollRsp rsp = { /*.consumerId = consumerId,*/ .numOfTopics = 0, @@ -288,62 +290,76 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { rsp.reqOffset = pReq->currentOffset; rsp.skipLogNum = 0; - SWalHead* pHead; while (1) { /*if (fetchOffset > walGetLastVer(pTq->pWal) || walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {*/ - if (walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) { + SWalReadHead* pHead; + if (walReadWithHandle_s(pTopic->pReadhandle, fetchOffset, &pHead) < 0) { // TODO: no more log, set timer to wait blocking time // if data inserted during waiting, launch query and // response to user break; } - int8_t pos = fetchOffset % TQ_BUFFER_SIZE; - pHead = pTopic->pReadhandle->pHead; - if (pHead->head.msgType == TDMT_VND_SUBMIT) { - SSubmitReq* pCont = (SSubmitReq*)&pHead->head.body; - qTaskInfo_t task = pTopic->buffer.output[pos].task; + /*printf("vg %d offset %ld msgType %d from epoch %d\n", pTq->pVnode->vgId, fetchOffset, pHead->msgType, + * pReq->epoch);*/ + /*int8_t pos = fetchOffset % TQ_BUFFER_SIZE;*/ + /*pHead = pTopic->pReadhandle->pHead;*/ + if (pHead->msgType == TDMT_VND_SUBMIT) { + SSubmitReq* pCont = (SSubmitReq*)&pHead->body; + printf("from topic %s from consumer\n", pTopic->topicName, consumerId); + qTaskInfo_t task = pTopic->buffer.output[workerId].task; + ASSERT(task); qSetStreamInput(task, pCont, STREAM_DATA_TYPE_SUBMIT_BLOCK); SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock)); while (1) { - SSDataBlock* pDataBlock; + SSDataBlock* pDataBlock = NULL; uint64_t ts; if (qExecTask(task, &pDataBlock, &ts) < 0) { ASSERT(false); } if (pDataBlock == NULL) { - fetchOffset++; - pos = fetchOffset % TQ_BUFFER_SIZE; - rsp.skipLogNum++; + /*pos = fetchOffset % TQ_BUFFER_SIZE;*/ break; } taosArrayPush(pRes, pDataBlock); - rsp.schema = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper; - rsp.rspOffset = fetchOffset; - - rsp.numOfTopics = 1; - rsp.pBlockData = pRes; - - int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqPollRsp(NULL, &rsp); - void* buf = rpcMallocCont(tlen); - if (buf == NULL) { - pMsg->code = -1; - return -1; - } - ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP; - ((SMqRspHead*)buf)->epoch = pReq->epoch; - ((SMqRspHead*)buf)->consumerId = consumerId; - - void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead)); - tEncodeSMqPollRsp(&abuf, &rsp); - /*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/ - pMsg->pCont = buf; - pMsg->contLen = tlen; - pMsg->code = 0; - tmsgSendRsp(pMsg); - return 0; } + + if (taosArrayGetSize(pRes) == 0) { + fetchOffset++; + rsp.skipLogNum++; + taosArrayDestroy(pRes); + continue; + } + rsp.schema = pTopic->buffer.output[workerId].pReadHandle->pSchemaWrapper; + rsp.rspOffset = fetchOffset; + + rsp.numOfTopics = 1; + rsp.pBlockData = pRes; + + int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqPollRsp(NULL, &rsp); + void* buf = rpcMallocCont(tlen); + if (buf == NULL) { + pMsg->code = -1; + taosMemoryFree(pHead); + return -1; + } + ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP; + ((SMqRspHead*)buf)->epoch = pReq->epoch; + ((SMqRspHead*)buf)->consumerId = consumerId; + + void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead)); + tEncodeSMqPollRsp(&abuf, &rsp); + /*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/ + pMsg->pCont = buf; + pMsg->contLen = tlen; + pMsg->code = 0; + printf("vg %d offset %ld msgType %d from epoch %d actual rsp\n", pTq->pVnode->vgId, fetchOffset, pHead->msgType, + pReq->epoch); + tmsgSendRsp(pMsg); + taosMemoryFree(pHead); + return 0; } else { + taosMemoryFree(pHead); fetchOffset++; rsp.skipLogNum++; } @@ -368,6 +384,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { pMsg->contLen = tlen; pMsg->code = 0; tmsgSendRsp(pMsg); + printf("vg %d offset %ld from epoch %d not rsp\n", pTq->pVnode->vgId, fetchOffset, pReq->epoch); /*}*/ return 0; @@ -432,7 +449,9 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) { }; pTopic->buffer.output[i].pReadHandle = pReadHandle; pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(req.qmsg, &handle); + ASSERT(pTopic->buffer.output[i].task); } + printf("set topic %s to consumer %ld\n", pTopic->topicName, req.consumerId); taosArrayPush(pConsumer->topics, pTopic); tqHandleMovePut(pTq->tqMeta, req.consumerId, pConsumer); tqHandleCommit(pTq->tqMeta, req.consumerId); diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index 9c0b0802ab..344b4fa655 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -167,7 +167,7 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) { if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) { break; } - if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) { + if (colDataAppend(pColData, curRow, sVal.val, false) < 0) { taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock); return NULL; } diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index 598647f797..053c03cc97 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -66,7 +66,7 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) { case TDMT_VND_TABLE_META: return vnodeGetTableMeta(pVnode, pMsg); case TDMT_VND_CONSUME: - return tqProcessPollReq(pVnode->pTq, pMsg); + return tqProcessPollReq(pVnode->pTq, pMsg, pInfo->workerId); case TDMT_VND_TASK_PIPE_EXEC: case TDMT_VND_TASK_MERGE_EXEC: return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, pInfo->workerId); diff --git a/source/dnode/vnode/src/vnd/vnodeWrite.c b/source/dnode/vnode/src/vnd/vnodeWrite.c index 8875090b43..631727fd5b 100644 --- a/source/dnode/vnode/src/vnd/vnodeWrite.c +++ b/source/dnode/vnode/src/vnd/vnodeWrite.c @@ -163,6 +163,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { // } break; case TDMT_VND_SUBMIT: + printf("vnode %d write data %ld\n", pVnode->vgId, ver); if (pVnode->config.streamMode == 0) { if (tsdbInsertData(pVnode->pTsdb, (SSubmitReq *)ptr, NULL) < 0) { // TODO: handle error diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 2df18f135c..d6c07b6659 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1551,8 +1551,8 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe if (pSDataBlock->pDataBlock != NULL) { SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, 0); tsCols = (int64_t*)pColDataInfo->pData; - assert(tsCols[0] == pSDataBlock->info.window.skey && - tsCols[pSDataBlock->info.rows - 1] == pSDataBlock->info.window.ekey); + /*assert(tsCols[0] == pSDataBlock->info.window.skey &&*/ + /*tsCols[pSDataBlock->info.rows - 1] == pSDataBlock->info.window.ekey);*/ } int32_t startPos = ascScan? 0 : (pSDataBlock->info.rows - 1); @@ -6994,6 +6994,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo *pOperator, bool* newgroup finalizeUpdatedResult(pInfo->binfo.pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf, pUpdated, pInfo->binfo.rowCellInfoOffset); blockDataEnsureCapacity(pInfo->binfo.pRes, pInfo->binfo.capacity); + initGroupResInfo(&pInfo->groupResInfo, &pInfo->binfo.resultRowInfo); toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pInfo->binfo.pRes, pInfo->binfo.capacity, pInfo->binfo.rowCellInfoOffset); diff --git a/source/libs/planner/src/planPhysiCreater.c b/source/libs/planner/src/planPhysiCreater.c index 9ef1c01cd1..d8a44e7156 100644 --- a/source/libs/planner/src/planPhysiCreater.c +++ b/source/libs/planner/src/planPhysiCreater.c @@ -719,6 +719,11 @@ static int32_t createStreamScanPhysiNodeByExchange(SPhysiPlanContext* pCxt, SExc if (NULL == pScan->pScanCols) { code = TSDB_CODE_OUT_OF_MEMORY; } + + if (TSDB_CODE_SUCCESS == code) { + code = sortScanCols(pScan->pScanCols); + } + if (TSDB_CODE_SUCCESS == code) { code = addDataBlockSlots(pCxt, pScan->pScanCols, pScan->node.pOutputDataBlockDesc); } diff --git a/source/libs/stream/src/tstream.c b/source/libs/stream/src/tstream.c index fb6c0f6c12..e2c5979a5a 100644 --- a/source/libs/stream/src/tstream.c +++ b/source/libs/stream/src/tstream.c @@ -38,7 +38,7 @@ static int32_t streamBuildDispatchMsg(SStreamTask* pTask, SArray* data, SRpcMsg* req.taskId = pTask->fixedEpDispatcher.taskId; } else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) { - // TODO fix tbname issue + // TODO use general name rule of schemaless char ctbName[TSDB_TABLE_FNAME_LEN + 22]; // all groupId must be the same in an array SSDataBlock* pBlock = taosArrayGet(data, 0); diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index 5296a16703..2545eeab4b 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -30,6 +30,9 @@ SWalReadHandle *walOpenReadHandle(SWal *pWal) { pRead->curFileFirstVer = -1; pRead->capacity = 0; pRead->status = 0; + + taosThreadMutexInit(&pRead->mutex, NULL); + pRead->pHead = taosMemoryMalloc(sizeof(SWalHead)); if (pRead->pHead == NULL) { terrno = TSDB_CODE_WAL_OUT_OF_MEMORY; @@ -135,6 +138,22 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) { return 0; } +int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead) { + taosThreadMutexLock(&pRead->mutex); + if (walReadWithHandle(pRead, ver) < 0) { + taosThreadMutexUnlock(&pRead->mutex); + return -1; + } + *ppHead = taosMemoryMalloc(sizeof(SWalReadHead) + pRead->pHead->head.len); + if (*ppHead == NULL) { + taosThreadMutexUnlock(&pRead->mutex); + return -1; + } + memcpy(*ppHead, &pRead->pHead->head, sizeof(SWalReadHead) + pRead->pHead->head.len); + taosThreadMutexUnlock(&pRead->mutex); + return 0; +} + int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) { int code; // TODO: check wal life @@ -145,7 +164,9 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) { } } - if (!taosValidFile(pRead->pReadLogTFile)) return -1; + if (!taosValidFile(pRead->pReadLogTFile)) { + return -1; + } code = taosReadFile(pRead->pReadLogTFile, pRead->pHead, sizeof(SWalHead)); if (code != sizeof(SWalHead)) { diff --git a/source/util/src/tjson.c b/source/util/src/tjson.c index 0efcf517a9..93a843fee8 100644 --- a/source/util/src/tjson.c +++ b/source/util/src/tjson.c @@ -121,7 +121,8 @@ int32_t tjsonAddItem(SJson* pJson, FToJson func, const void* pObj) { return tjsonAddItemToArray(pJson, pJobj); } -int32_t tjsonAddArray(SJson* pJson, const char* pName, FToJson func, const void* pArray, int32_t itemSize, int32_t num) { +int32_t tjsonAddArray(SJson* pJson, const char* pName, FToJson func, const void* pArray, int32_t itemSize, + int32_t num) { if (num > 0) { SJson* pJsonArray = tjsonAddArrayToObject(pJson, pName); if (NULL == pJsonArray) { @@ -168,7 +169,7 @@ int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal } *pVal = strtol(p, NULL, 10); - return (errno == EINVAL || errno == ERANGE) ? TSDB_CODE_FAILED:TSDB_CODE_SUCCESS; + return TSDB_CODE_SUCCESS; } int32_t tjsonGetIntValue(const SJson* pJson, const char* pName, int32_t* pVal) { @@ -199,19 +200,19 @@ int32_t tjsonGetUBigIntValue(const SJson* pJson, const char* pName, uint64_t* pV } *pVal = strtoul(p, NULL, 10); - return (errno == ERANGE||errno == EINVAL) ? TSDB_CODE_FAILED:TSDB_CODE_SUCCESS; + return TSDB_CODE_SUCCESS; } int32_t tjsonGetUIntValue(const SJson* pJson, const char* pName, uint32_t* pVal) { uint64_t val = 0; - int32_t code = tjsonGetUBigIntValue(pJson, pName, &val); + int32_t code = tjsonGetUBigIntValue(pJson, pName, &val); *pVal = val; return code; } int32_t tjsonGetUTinyIntValue(const SJson* pJson, const char* pName, uint8_t* pVal) { uint64_t val = 0; - int32_t code = tjsonGetUBigIntValue(pJson, pName, &val); + int32_t code = tjsonGetUBigIntValue(pJson, pName, &val); *pVal = val; return code; } @@ -264,7 +265,7 @@ int32_t tjsonMakeObject(const SJson* pJson, const char* pName, FToObject func, v int32_t tjsonToArray(const SJson* pJson, const char* pName, FToObject func, void* pArray, int32_t itemSize) { const cJSON* jArray = tjsonGetObjectItem(pJson, pName); - int32_t size = (NULL == jArray ? 0 : tjsonGetArraySize(jArray)); + int32_t size = (NULL == jArray ? 0 : tjsonGetArraySize(jArray)); for (int32_t i = 0; i < size; ++i) { int32_t code = func(tjsonGetArrayItem(jArray, i), (char*)pArray + itemSize * i); if (TSDB_CODE_SUCCESS != code) { diff --git a/tests/script/tsim/tmq/basic1.sim b/tests/script/tsim/tmq/basic1.sim index fe6a7a0660..f84d70f711 100644 --- a/tests/script/tsim/tmq/basic1.sim +++ b/tests/script/tsim/tmq/basic1.sim @@ -35,7 +35,7 @@ sql connect $dbNamme = d0 print =============== create database , vgroup 1 -sql create database $dbNamme vgroups 1 +sql create database $dbNamme vgroups 2 sql show databases print $data00 $data01 $data02 if $rows != 2 then diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index 4d3108500e..38264331c1 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -226,7 +226,7 @@ void loop_consume(tmq_t* tmq) { int32_t totalRows = 0; int32_t skipLogNum = 0; while (running) { - tmq_message_t* tmqMsg = tmq_consumer_poll(tmq, 1); + tmq_message_t* tmqMsg = tmq_consumer_poll(tmq, 3000); if (tmqMsg) { totalMsgs++;