xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix

This commit is contained in:
Liu Jicong 2022-03-31 15:52:54 +08:00
parent 915c9f67fe
commit 3e5da8bedb
20 changed files with 240 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
example/src/tmq.c
View File

@ -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);*/

4
example/src/tstream.c
View File

@ -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;
}

29
include/common/tcommon.h
View File

@ -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)

78
include/common/trow.h
View File

@ -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 {

34
include/libs/wal/wal.h
View File

@ -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

15
source/client/src/tmq.c
View File

@ -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++;

3
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -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;

2
source/dnode/vnode/src/inc/vnd.h
View File

@ -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);

2
source/dnode/vnode/src/meta/metaTbUid.c
View File

@ -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();
}

91
source/dnode/vnode/src/tq/tq.c
View File

@ -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);

2
source/dnode/vnode/src/tq/tqRead.c
View File

@ -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;
}

2
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -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);

1
source/dnode/vnode/src/vnd/vnodeWrite.c
View File

@ -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

5
source/libs/executor/src/executorimpl.c
View File

@ -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);

5
source/libs/planner/src/planPhysiCreater.c
View File

@ -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);
}

2
source/libs/stream/src/tstream.c
View File

@ -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);

23
source/libs/wal/src/walRead.c
View File

@ -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)) {

13
source/util/src/tjson.c
View File

@ -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) {

2
tests/script/tsim/tmq/basic1.sim
View File

@ -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

2
tests/test/c/tmqSim.c
View File

@ -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++;