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

Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/commit_txn

This commit is contained in:
Hongze Cheng 2022-10-19 09:02:01 +08:00
parent 74d025c783 b621f04e40
commit 864111d85c
10 changed files with 58 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/taostools_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG c64858f
GIT_TAG 9284147
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE

10
source/client/src/clientSml.c
View File

@ -1372,8 +1372,14 @@ static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
SHashObj *s1 = *(SHashObj **)key1;
SHashObj *s2 = *(SHashObj **)key2;
SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
SSmlKv **kv1pp = (SSmlKv **)taosHashGet(s1, TS, TS_LEN);
SSmlKv **kv2pp = (SSmlKv **)taosHashGet(s2, TS, TS_LEN);
if(!kv1pp || !kv2pp){
uError("smlKvTimeHashCompare kv is null");
return -1;
}
SSmlKv *kv1 = *kv1pp;
SSmlKv *kv2 = *kv2pp;
if(!kv1 || kv1->type != TSDB_DATA_TYPE_TIMESTAMP){
uError("smlKvTimeHashCompare kv1");
return -1;

49
source/client/src/clientTmq.c
View File

@ -210,6 +210,8 @@ typedef struct {
typedef struct {
SMqCommitCbParamSet* params;
STqOffset* pOffset;
/*char topicName[TSDB_TOPIC_FNAME_LEN];*/
/*int32_t vgId;*/
} SMqCommitCbParam;
tmq_conf_t* tmq_conf_new() {
@ -407,6 +409,14 @@ int32_t tmqCommitDone(SMqCommitCbParamSet* pParamSet) {
return 0;
}
static void tmqCommitRspCountDown(SMqCommitCbParamSet* pParamSet) {
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
}
int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
SMqCommitCbParamSet* pParamSet = (SMqCommitCbParamSet*)pParam->params;
@ -420,18 +430,13 @@ int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
#endif
taosMemoryFree(pParam->pOffset);
if (pBuf->pData) taosMemoryFree(pBuf->pData);
taosMemoryFree(pBuf->pData);
/*tscDebug("receive offset commit cb of %s on vgId:%d, offset is %" PRId64, pParam->pOffset->subKey, pParam->->vgId,
* pOffset->version);*/
// count down waiting rsp
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
tmqCommitRspCountDown(pParamSet);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
return 0;
}
@ -591,14 +596,10 @@ FAIL:
return 0;
}
int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
static int32_t tmqCommitConsumerImpl(tmq_t* tmq, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
void* userParam) {
int32_t code = -1;
if (msg != NULL) {
return tmqCommitMsgImpl(tmq, msg, async, userCb, userParam);
}
SMqCommitCbParamSet* pParamSet = taosMemoryCalloc(1, sizeof(SMqCommitCbParamSet));
if (pParamSet == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
@ -646,33 +647,37 @@ int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t
}
}
// no request is sent
if (pParamSet->totalRspNum == 0) {
tsem_destroy(&pParamSet->rspSem);
taosMemoryFree(pParamSet);
return 0;
}
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
// count down since waiting rsp num init as 1
tmqCommitRspCountDown(pParamSet);
if (!async) {
tsem_wait(&pParamSet->rspSem);
code = pParamSet->rspErr;
tsem_destroy(&pParamSet->rspSem);
taosMemoryFree(pParamSet);
}
#if 0
if (!async) {
taosArrayDestroyP(pParamSet->successfulOffsets, taosMemoryFree);
taosArrayDestroyP(pParamSet->failedOffsets, taosMemoryFree);
}
#endif
}
return 0;
return code;
}
int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
void* userParam) {
if (msg) {
return tmqCommitMsgImpl(tmq, msg, async, userCb, userParam);
} else {
return tmqCommitConsumerImpl(tmq, automatic, async, userCb, userParam);
}
}
void tmqAssignAskEpTask(void* param, void* tmrId) {

6
source/dnode/vnode/src/inc/tq.h
View File

@ -149,7 +149,7 @@ int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
// tqRead
int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset);
int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
@ -181,8 +181,8 @@ int32_t tqOffsetDelete(STqOffsetStore* pStore, const char* subscribeKey)
int32_t tqOffsetCommitFile(STqOffsetStore* pStore);
// tqSink
void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
// tqOffset
char* tqOffsetBuildFName(const char* path, int32_t ver);

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

@ -595,7 +595,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
tqInitTaosxRsp(&taosxRsp, pReq);
if (fetchOffsetNew.type != TMQ_OFFSET__LOG) {
tqScan(pTq, pHandle, &taosxRsp, &metaRsp, &fetchOffsetNew);
tqScanTaosx(pTq, pHandle, &taosxRsp, &metaRsp, &fetchOffsetNew);
if (metaRsp.metaRspLen > 0) {
if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
@ -924,7 +924,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask) {
pTask->smaSink.smaSink = smaHandleRes;
} else if (pTask->outputType == TASK_OUTPUT__TABLE) {
pTask->tbSink.vnode = pTq->pVnode;
pTask->tbSink.tbSinkFunc = tqTableSink1;
pTask->tbSink.tbSinkFunc = tqSinkToTablePipeline;
ASSERT(pTask->tbSink.pSchemaWrapper);
ASSERT(pTask->tbSink.pSchemaWrapper->pSchema);

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

@ -123,7 +123,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
return 0;
}
int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
const STqExecHandle* pExec = &pHandle->execHandle;
qTaskInfo_t task = pExec->task;

4
source/dnode/vnode/src/tq/tqSink.c
View File

@ -284,7 +284,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
return ret;
}
void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
const SArray* pBlocks = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
int64_t suid = pTask->tbSink.stbUid;
@ -528,7 +528,7 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
taosArrayDestroy(tagArray);
}
void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
const SArray* pRes = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
SBatchDeleteReq deleteReq = {0};

141
utils/test/c/sml_test.c
View File

@ -102,27 +102,8 @@ int smlProcess_json1_Test() {
taos_free_result(pRes);
const char *sql[] = {
"["
" {"
" \"metric\": \"sys.cpu.nice\","
" \"timestamp\": 0,"
" \"value\": 18,"
" \"tags\": {"
" \"host\": \"web01\","
" \"id\": \"t1\","
" \"dc\": \"lga\""
" }"
" },"
" {"
" \"metric\": \"sys.cpu.nice\","
" \"timestamp\": 1662344042,"
" \"value\": 9,"
" \"tags\": {"
" \"host\": \"web02\","
" \"dc\": \"lga\""
" }"
" }"
"]",};
"[{\"metric\":\"sys.cpu.nice\",\"timestamp\":0,\"value\":18,\"tags\":{\"host\":\"web01\",\"id\":\"t1\",\"dc\":\"lga\"}},{\"metric\":\"sys.cpu.nice\",\"timestamp\":1662344042,\"value\":9,\"tags\":{\"host\":\"web02\",\"dc\":\"lga\"}}]"
};
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
TSDB_SML_TIMESTAMP_NANO_SECONDS);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
@ -143,28 +124,8 @@ int smlProcess_json2_Test() {
taos_free_result(pRes);
const char *sql[] = {
"{"
" \"metric\": \"meter_current0\","
" \"timestamp\": {"
" \"value\" : 1662344042,"
" \"type\" : \"s\""
" },"
" \"value\": {"
" \"value\" : 10.3,"
" \"type\" : \"i64\""
" },"
" \"tags\": {"
" \"groupid\": { "
" \"value\" : 2,"
" \"type\" : \"bigint\""
" },"
" \"location\": { "
" \"value\" : \"北京\","
" \"type\" : \"binary\""
" },"
" \"id\": \"d1001\""
" }"
"}",};
"{\"metric\":\"meter_current0\",\"timestamp\":{\"value\":1662344042,\"type\":\"s\"},\"value\":{\"value\":10.3,\"type\":\"i64\"},\"tags\":{\"groupid\":{\"value\":2,\"type\":\"bigint\"},\"location\":{\"value\":\"北京\",\"type\":\"binary\"},\"id\":\"d1001\"}}"
};
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
TSDB_SML_TIMESTAMP_NANO_SECONDS);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
@ -185,56 +146,7 @@ int smlProcess_json3_Test() {
taos_free_result(pRes);
const char *sql[] = {
"{"
" \"metric\": \"meter_current1\","
" \"timestamp\": {"
" \"value\" : 1662344042,"
" \"type\" : \"s\""
" },"
" \"value\": {"
" \"value\" : 10.3,"
" \"type\" : \"i64\""
" },"
" \"tags\": {"
" \"t1\": { "
" \"value\" : 2,"
" \"type\" : \"bigint\""
" },"
" \"t2\": { "
" \"value\" : 2,"
" \"type\" : \"int\""
" },"
" \"t3\": { "
" \"value\" : 2,"
" \"type\" : \"i16\""
" },"
" \"t4\": { "
" \"value\" : 2,"
" \"type\" : \"i8\""
" },"
" \"t5\": { "
" \"value\" : 2,"
" \"type\" : \"f32\""
" },"
" \"t6\": { "
" \"value\" : 2,"
" \"type\" : \"double\""
" },"
" \"t7\": { "
" \"value\" : \"8323\","
" \"type\" : \"binary\""
" },"
" \"t8\": { "
" \"value\" : \"北京\","
" \"type\" : \"nchar\""
" },"
" \"t9\": { "
" \"value\" : true,"
" \"type\" : \"bool\""
" },"
" \"id\": \"d1001\""
" }"
"}",};
"{\"metric\":\"meter_current1\",\"timestamp\":{\"value\":1662344042,\"type\":\"s\"},\"value\":{\"value\":10.3,\"type\":\"i64\"},\"tags\":{\"t1\":{\"value\":2,\"type\":\"bigint\"},\"t2\":{\"value\":2,\"type\":\"int\"},\"t3\":{\"value\":2,\"type\":\"i16\"},\"t4\":{\"value\":2,\"type\":\"i8\"},\"t5\":{\"value\":2,\"type\":\"f32\"},\"t6\":{\"value\":2,\"type\":\"double\"},\"t7\":{\"value\":\"8323\",\"type\":\"binary\"},\"t8\":{\"value\":\"北京\",\"type\":\"nchar\"},\"t9\":{\"value\":true,\"type\":\"bool\"},\"id\":\"d1001\"}}"};
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
TSDB_SML_TIMESTAMP_NANO_SECONDS);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
@ -255,47 +167,8 @@ int smlProcess_json4_Test() {
taos_free_result(pRes);
const char *sql[] = {
"{"
" \"metric\": \"meter_current2\","
" \"timestamp\": {"
" \"value\" : 1662344042000,"
" \"type\" : \"ms\""
" },"
" \"value\": \"ni\","
" \"tags\": {"
" \"t1\": { "
" \"value\" : 20,"
" \"type\" : \"i64\""
" },"
" \"t2\": { "
" \"value\" : 25,"
" \"type\" : \"i32\""
" },"
" \"t3\": { "
" \"value\" : 2,"
" \"type\" : \"smallint\""
" },"
" \"t4\": { "
" \"value\" : 2,"
" \"type\" : \"tinyint\""
" },"
" \"t5\": { "
" \"value\" : 2,"
" \"type\" : \"float\""
" },"
" \"t6\": { "
" \"value\" : 0.2,"
" \"type\" : \"f64\""
" },"
" \"t7\": \"nsj\","
" \"t8\": { "
" \"value\" : \"北京\","
" \"type\" : \"nchar\""
" },"
" \"t9\": false,"
" \"id\": \"d1001\""
" }"
"}",};
"{\"metric\":\"meter_current2\",\"timestamp\":{\"value\":1662344042000,\"type\":\"ms\"},\"value\":\"ni\",\"tags\":{\"t1\":{\"value\":20,\"type\":\"i64\"},\"t2\":{\"value\":25,\"type\":\"i32\"},\"t3\":{\"value\":2,\"type\":\"smallint\"},\"t4\":{\"value\":2,\"type\":\"tinyint\"},\"t5\":{\"value\":2,\"type\":\"float\"},\"t6\":{\"value\":0.2,\"type\":\"f64\"},\"t7\":\"nsj\",\"t8\":{\"value\":\"北京\",\"type\":\"nchar\"},\"t9\":false,\"id\":\"d1001\"}}"
};
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
TSDB_SML_TIMESTAMP_NANO_SECONDS);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));

10
utils/test/c/tmqDemo.c
View File

@ -130,15 +130,15 @@ void parseArgument(int32_t argc, char* argv[]) {
printHelp();
exit(0);
} else if (strcmp(argv[i], "-d") == 0) {
strcpy(g_stConfInfo.dbName, argv[++i]);
tstrncpy(g_stConfInfo.dbName, argv[++i], sizeof(g_stConfInfo.dbName));
} else if (strcmp(argv[i], "-c") == 0) {
strcpy(configDir, argv[++i]);
tstrncpy(configDir, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-s") == 0) {
strcpy(g_stConfInfo.stbName, argv[++i]);
tstrncpy(g_stConfInfo.stbName, argv[++i], sizeof(g_stConfInfo.stbName));
} else if (strcmp(argv[i], "-w") == 0) {
strcpy(g_stConfInfo.vnodeWalPath, argv[++i]);
tstrncpy(g_stConfInfo.vnodeWalPath, argv[++i], sizeof(g_stConfInfo.vnodeWalPath));
} else if (strcmp(argv[i], "-f") == 0) {
strcpy(g_stConfInfo.resultFileName, argv[++i]);
tstrncpy(g_stConfInfo.resultFileName, argv[++i], sizeof(g_stConfInfo.resultFileName));
} else if (strcmp(argv[i], "-t") == 0) {
g_stConfInfo.numOfThreads = atoi(argv[++i]);
} else if (strcmp(argv[i], "-n") == 0) {

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

@ -949,7 +949,7 @@ void parseConsumeInfo() {
token = strtok(g_stConfInfo.stThreads[i].topicString, delim);
while (token != NULL) {
// printf("%s\n", token );
strcpy(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic], token);
tstrncpy(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic], token, sizeof(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic]));
ltrim(g_stConfInfo.stThreads[i].topics[g_stConfInfo.stThreads[i].numOfTopic]);
// printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]);
g_stConfInfo.stThreads[i].numOfTopic++;