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

merge 3.0

This commit is contained in:
Xiaoyu Wang 2023-02-27 14:16:33 +08:00
parent d6dd087f84 c5348ce793
commit f97a140a3c
25 changed files with 287 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Jenkinsfile2
View File

@ -387,7 +387,7 @@ pipeline {
}
steps {
catchError(buildResult: 'FAILURE', stageResult: 'FAILURE') {
timeout(time: 75, unit: 'MINUTES'){
timeout(time: 126, unit: 'MINUTES'){
pre_test_win()
pre_test_build_win()
run_win_ctest()
@ -423,7 +423,7 @@ pipeline {
echo "${WKDIR}/restore.sh -p ${BRANCH_NAME} -n ${BUILD_ID} -c {container name}"
}
catchError(buildResult: 'FAILURE', stageResult: 'FAILURE') {
timeout(time: 120, unit: 'MINUTES'){
timeout(time: 130, unit: 'MINUTES'){
pre_test()
script {
sh '''

7
examples/JDBC/springbootdemo/readme.md
View File

@ -1,6 +1,13 @@
## TDengine SpringBoot + Mybatis Demo
## 需要提前创建 test 数据库
```
$ taos -s 'create database if not exists test'
$ curl http://localhost:8080/weather/init
```
### 配置 application.properties
```properties
# datasource config

10
examples/JDBC/springbootdemo/src/main/java/com/taosdata/example/springbootdemo/dao/WeatherMapper.xml
View File

@ -7,6 +7,7 @@
<id column="ts" jdbcType="TIMESTAMP" property="ts"/>
<result column="temperature" jdbcType="FLOAT" property="temperature"/>
<result column="humidity" jdbcType="FLOAT" property="humidity"/>
<result column="bytes" jdbcType="BINARY" property="bytes" />
</resultMap>
<select id="lastOne" resultType="java.util.Map">
@ -36,6 +37,11 @@
binary
(
64
),
bytes
binary
(
64
)) tags
(
location nchar
@ -63,8 +69,8 @@
</select>
<insert id="insert" parameterType="com.taosdata.example.springbootdemo.domain.Weather">
insert into test.t#{groupId} (ts, temperature, humidity, note)
values (#{ts}, ${temperature}, ${humidity}, #{note})
insert into test.t#{groupId} (ts, temperature, humidity, note, bytes)
values (#{ts}, ${temperature}, ${humidity}, #{note}, #{bytes})
</insert>
<select id="getSubTables" resultType="String">

30
examples/JDBC/springbootdemo/src/main/java/com/taosdata/example/springbootdemo/domain/Weather.java
View File

@ -2,6 +2,7 @@ package com.taosdata.example.springbootdemo.domain;
import com.fasterxml.jackson.annotation.JsonFormat;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
public class Weather {
@ -12,6 +13,9 @@ public class Weather {
private Float humidity;
private String location;
private String note;
// In rest mode, the byte[] type is not recommended.
// UTF-8 is used to encode the byte arrays, that result may affect the SQL correctness
private byte[] bytes;
private int groupId;
public Weather() {
@ -70,4 +74,30 @@ public class Weather {
public void setNote(String note) {
this.note = note;
}
public byte[] getBytes() {
return bytes;
}
public void setBytes(byte[] bytes) {
this.bytes = bytes;
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("Weather{");
sb.append("ts=").append(ts);
sb.append(", temperature=").append(temperature);
sb.append(", humidity=").append(humidity);
sb.append(", location='").append(location).append('\'');
sb.append(", note='").append(note).append('\'');
sb.append(", bytes -> string=");
if (bytes == null) sb.append("null");
else {
sb.append(new String(bytes, StandardCharsets.UTF_8));
}
sb.append(", groupId=").append(groupId);
sb.append('}');
return sb.toString();
}
}

2
examples/JDBC/springbootdemo/src/main/java/com/taosdata/example/springbootdemo/service/WeatherService.java
View File

@ -5,6 +5,7 @@ import com.taosdata.example.springbootdemo.domain.Weather;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.List;
import java.util.Map;
@ -30,6 +31,7 @@ public class WeatherService {
weather.setLocation(locations[random.nextInt(locations.length)]);
weather.setGroupId(i % locations.length);
weather.setNote("note-" + i);
weather.setBytes(locations[random.nextInt(locations.length)].getBytes(StandardCharsets.UTF_8));
weatherMapper.createTable(weather);
count += weatherMapper.insert(weather);
}

4
include/common/ttypes.h
View File

@ -53,10 +53,6 @@ typedef struct {
#define varDataNetLen(v) (htons(((VarDataLenT *)(v))[0]))
#define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v))
// this data type is internally used only in 'in' query to hold the values
#define TSDB_DATA_TYPE_POINTER_ARRAY (1000)
#define TSDB_DATA_TYPE_VALUE_ARRAY (1001)
#define GET_TYPED_DATA(_v, _finalType, _type, _data) \
do { \
switch (_type) { \

4
include/libs/transport/thttp.h
View File

@ -17,6 +17,7 @@
#define _TD_UTIL_HTTP_H_
#include "os.h"
#include "tref.h"
#ifdef __cplusplus
extern "C" {
@ -24,7 +25,8 @@ extern "C" {
typedef enum { HTTP_GZIP, HTTP_FLAT } EHttpCompFlag;
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag);
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag);
#ifdef __cplusplus
}

5
include/os/osMemory.h
View File

@ -29,9 +29,12 @@ extern "C" {
#define calloc CALLOC_FUNC_TAOS_FORBID
#define realloc REALLOC_FUNC_TAOS_FORBID
#define free FREE_FUNC_TAOS_FORBID
#ifdef strdup
#undef strdup
#define strdup STRDUP_FUNC_TAOS_FORBID
#endif // ifndef ALLOW_FORBID_FUNC
#endif
#endif // ifndef ALLOW_FORBID_FUNC
#endif // if !defined(WINDOWS)
int32_t taosMemoryDbgInit();

3
source/client/jni/com_taosdata_jdbc_tmq_TMQConnector.h
View File

@ -99,6 +99,9 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitSync(JNI
*/
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitAsync(JNIEnv *, jobject, jlong, jlong, jobject);
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_consumerCommitAsync(JNIEnv *, jobject, jlong, jlong,
jobject);
/*
* Class: com_taosdata_jdbc_tmq_TMQConnector
* Method: tmqUnsubscribeImp

1
source/client/src/clientJniConnector.c
View File

@ -56,6 +56,7 @@ jmethodID g_createConsumerErrorCallback;
jmethodID g_topicListCallback;
jclass g_consumerClass;
// deprecated
jmethodID g_commitCallback;
void jniGetGlobalMethod(JNIEnv *env) {

66
source/client/src/clientTmqConnector.c
View File

@ -17,6 +17,36 @@
#include "jniCommon.h"
#include "taos.h"
int __init_tmq = 0;
jmethodID g_offsetCallback;
void tmqGlobalMethod(JNIEnv *env) {
// make sure init function executed once
switch (atomic_val_compare_exchange_32(&__init_tmq, 0, 1)) {
case 0:
break;
case 1:
do {
taosMsleep(0);
} while (atomic_load_32(&__init_tmq) == 1);
case 2:
return;
}
if (g_vm == NULL) {
(*env)->GetJavaVM(env, &g_vm);
}
jclass offset = (*env)->FindClass(env, "com/taosdata/jdbc/tmq/OffsetWaitCallback");
jclass g_offsetCallbackClass = (*env)->NewGlobalRef(env, offset);
g_offsetCallback = (*env)->GetMethodID(env, g_offsetCallbackClass, "commitCallbackHandler", "(I)V");
(*env)->DeleteLocalRef(env, offset);
atomic_store_32(&__init_tmq, 2);
jniDebug("tmq method register finished");
}
// deprecated
void commit_cb(tmq_t *tmq, int32_t code, void *param) {
JNIEnv *env = NULL;
int status = (*g_vm)->GetEnv(g_vm, (void **)&env, JNI_VERSION_1_6);
@ -40,6 +70,28 @@ void commit_cb(tmq_t *tmq, int32_t code, void *param) {
env = NULL;
}
void consumer_callback(tmq_t *tmq, int32_t code, void *param) {
JNIEnv *env = NULL;
int status = (*g_vm)->GetEnv(g_vm, (void **)&env, JNI_VERSION_1_6);
bool needDetach = false;
if (status < 0) {
if ((*g_vm)->AttachCurrentThread(g_vm, (void **)&env, NULL) != 0) {
return;
}
needDetach = true;
}
jobject obj = (jobject)param;
(*env)->CallVoidMethod(env, obj, g_offsetCallback, code);
(*env)->DeleteGlobalRef(env, obj);
param = NULL;
if (needDetach) {
(*g_vm)->DetachCurrentThread(g_vm);
}
env = NULL;
}
JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqConfNewImp(JNIEnv *env, jobject jobj) {
tmq_conf_t *conf = tmq_conf_new();
jniGetGlobalMethod(env);
@ -201,6 +253,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitSync(JNI
return tmq_commit_sync(tmq, res);
}
// deprecated
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitAsync(JNIEnv *env, jobject jobj, jlong jtmq,
jlong jres, jobject consumer) {
tmq_t *tmq = (tmq_t *)jtmq;
@ -213,6 +266,19 @@ JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitAsync(JN
tmq_commit_async(tmq, res, commit_cb, consumer);
}
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_consumerCommitAsync(JNIEnv *env, jobject jobj, jlong jtmq,
jlong jres, jobject offset) {
tmqGlobalMethod(env);
tmq_t *tmq = (tmq_t *)jtmq;
if (tmq == NULL) {
jniError("jobj:%p, tmq is closed", jobj);
return;
}
TAOS_RES *res = (TAOS_RES *)jres;
offset = (*env)->NewGlobalRef(env, offset);
tmq_commit_async(tmq, res, consumer_callback, offset);
}
JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqUnsubscribeImp(JNIEnv *env, jobject jobj,
jlong jtmq) {
tmq_t *tmq = (tmq_t *)jtmq;

33
source/common/src/tvariant.c
View File

@ -145,19 +145,6 @@ void taosVariantDestroy(SVariant *pVar) {
pVar->nLen = 0;
}
// NOTE: this is only for string array
if (pVar->nType == TSDB_DATA_TYPE_POINTER_ARRAY) {
size_t num = taosArrayGetSize(pVar->arr);
for (size_t i = 0; i < num; i++) {
void *p = taosArrayGetP(pVar->arr, i);
taosMemoryFree(p);
}
taosArrayDestroy(pVar->arr);
pVar->arr = NULL;
} else if (pVar->nType == TSDB_DATA_TYPE_VALUE_ARRAY) {
taosArrayDestroy(pVar->arr);
pVar->arr = NULL;
}
}
void taosVariantAssign(SVariant *pDst, const SVariant *pSrc) {
@ -180,28 +167,8 @@ void taosVariantAssign(SVariant *pDst, const SVariant *pSrc) {
if (IS_NUMERIC_TYPE(pSrc->nType) || (pSrc->nType == TSDB_DATA_TYPE_BOOL)) {
pDst->i = pSrc->i;
} else if (pSrc->nType == TSDB_DATA_TYPE_POINTER_ARRAY) { // this is only for string array
size_t num = taosArrayGetSize(pSrc->arr);
pDst->arr = taosArrayInit(num, sizeof(char *));
for (size_t i = 0; i < num; i++) {
char *p = (char *)taosArrayGetP(pSrc->arr, i);
char *n = taosStrdup(p);
taosArrayPush(pDst->arr, &n);
}
} else if (pSrc->nType == TSDB_DATA_TYPE_VALUE_ARRAY) {
size_t num = taosArrayGetSize(pSrc->arr);
pDst->arr = taosArrayInit(num, sizeof(int64_t));
pDst->nLen = pSrc->nLen;
ASSERT(pSrc->nLen == num);
for (size_t i = 0; i < num; i++) {
int64_t *p = taosArrayGet(pSrc->arr, i);
taosArrayPush(pDst->arr, p);
}
}
if (pDst->nType != TSDB_DATA_TYPE_POINTER_ARRAY && pDst->nType != TSDB_DATA_TYPE_VALUE_ARRAY) {
pDst->nLen = tDataTypes[pDst->nType].bytes;
}
}
int32_t taosVariantCompare(const SVariant *p1, const SVariant *p2) {

12
source/dnode/mgmt/node_util/inc/dmUtil.h
View File

@ -53,12 +53,12 @@ extern "C" {
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
#define dGFatal(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dFatal(param ", gtid:%s", __VA_ARGS__, buf);}
#define dGError(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dError(param ", gtid:%s", __VA_ARGS__, buf);}
#define dGWarn(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dWarn (param ", gtid:%s", __VA_ARGS__, buf);}
#define dGInfo(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dInfo (param ", gtid:%s", __VA_ARGS__, buf);}
#define dGDebug(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dDebug(param ", gtid:%s", __VA_ARGS__, buf);}
#define dGTrace(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dTrace(param ", gtid:%s", __VA_ARGS__, buf);}
#define dGFatal(param, ...) {if (dDebugFlag & DEBUG_FATAL) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dFatal(param ", gtid:%s", __VA_ARGS__, buf);}}
#define dGError(param, ...) {if (dDebugFlag & DEBUG_ERROR) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dError(param ", gtid:%s", __VA_ARGS__, buf);}}
#define dGWarn(param, ...) {if (dDebugFlag & DEBUG_WARN) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dWarn(param ", gtid:%s", __VA_ARGS__, buf);}}
#define dGInfo(param, ...) {if (dDebugFlag & DEBUG_INFO) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dInfo(param ", gtid:%s", __VA_ARGS__, buf);}}
#define dGDebug(param, ...) {if (dDebugFlag & DEBUG_DEBUG) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dDebug(param ", gtid:%s", __VA_ARGS__, buf);}}
#define dGTrace(param, ...) {if (dDebugFlag & DEBUG_TRACE) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); dTrace(param ", gtid:%s", __VA_ARGS__, buf);}}
// clang-format on

12
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -41,12 +41,12 @@ extern "C" {
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
#define mGFatal(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mFatal(param ", gtid:%s", __VA_ARGS__, buf);}
#define mGError(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mError(param ", gtid:%s", __VA_ARGS__, buf);}
#define mGWarn(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mWarn (param ", gtid:%s", __VA_ARGS__, buf);}
#define mGInfo(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mInfo (param ", gtid:%s", __VA_ARGS__, buf);}
#define mGDebug(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mDebug(param ", gtid:%s", __VA_ARGS__, buf);}
#define mGTrace(param, ...) { char buf[40] = {0}; TRACE_TO_STR(trace, buf); mTrace(param ", gtid:%s", __VA_ARGS__, buf);}
#define mGFatal(param, ...) { if (mDebugFlag & DEBUG_FATAL){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mFatal(param ", gtid:%s", __VA_ARGS__, buf);}}
#define mGError(param, ...) { if (mDebugFlag & DEBUG_ERROR){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mError(param ", gtid:%s", __VA_ARGS__, buf);}}
#define mGWarn(param, ...) { if (mDebugFlag & DEBUG_WARN){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mWarn (param ", gtid:%s", __VA_ARGS__, buf);}}
#define mGInfo(param, ...) { if (mDebugFlag & DEBUG_INFO){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mInfo (param ", gtid:%s", __VA_ARGS__, buf);}}
#define mGDebug(param, ...) { if (mDebugFlag & DEBUG_DEBUG){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mDebug(param ", gtid:%s", __VA_ARGS__, buf);}}
#define mGTrace(param, ...) { if (mDebugFlag & DEBUG_TRACE){ char buf[40] = {0}; TRACE_TO_STR(trace, buf); mTrace(param ", gtid:%s", __VA_ARGS__, buf);}}
// clang-format on
#define SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE)

4
source/dnode/vnode/src/meta/metaTable.c
View File

@ -1921,10 +1921,10 @@ int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_
// refactor
if (IS_VAR_DATA_TYPE(type)) {
memcpy((*ppTagIdxKey)->data, (uint16_t *)&nTagData, VARSTR_HEADER_SIZE);
memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData);
if (pTagData != NULL) memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData);
*(tb_uid_t *)((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE + nTagData) = uid;
} else {
memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
if (pTagData != NULL) memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
*(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid;
}

18
source/libs/index/src/indexFstFile.c
View File

@ -18,6 +18,7 @@
#include "indexInt.h"
#include "indexUtil.h"
#include "os.h"
#include "osDef.h"
#include "tutil.h"
static int32_t kBlockSize = 4096;
@ -172,7 +173,8 @@ static FORCE_INLINE int idxFileCtxDoFlush(IFileCtx* ctx) {
int32_t nw = taosWriteFile(ctx->file.pFile, ctx->file.wBuf, ctx->file.wBufOffset);
ctx->file.wBufOffset = 0;
}
taosFsyncFile(ctx->file.pFile);
int ret = taosFsyncFile(ctx->file.pFile);
UNUSED(ret);
} else {
// do nothing
}
@ -180,11 +182,11 @@ static FORCE_INLINE int idxFileCtxDoFlush(IFileCtx* ctx) {
}
IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity) {
int code = 0;
IFileCtx* ctx = taosMemoryCalloc(1, sizeof(IFileCtx));
if (ctx == NULL) {
return NULL;
}
ctx->type = type;
if (ctx->type == TFILE) {
// ugly code, refactor later
@ -192,15 +194,21 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
memcpy(ctx->file.buf, path, strlen(path));
if (readOnly == false) {
ctx->file.pFile = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
taosFtruncateFile(ctx->file.pFile, 0);
taosStatFile(path, &ctx->file.size, NULL);
code = taosFtruncateFile(ctx->file.pFile, 0);
UNUSED(code);
code = taosStatFile(path, &ctx->file.size, NULL);
UNUSED(code);
ctx->file.wBufOffset = 0;
ctx->file.wBufCap = kBlockSize * 4;
ctx->file.wBuf = taosMemoryCalloc(1, ctx->file.wBufCap);
} else {
ctx->file.pFile = taosOpenFile(path, TD_FILE_READ);
taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL);
code = taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL);
UNUSED(code);
ctx->file.wBufOffset = 0;
#ifdef USE_MMAP

3
source/libs/parser/src/parTranslater.c
View File

@ -3342,6 +3342,9 @@ static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
if (TSDB_CODE_SUCCESS == code) {
code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange);
}
if (TSDB_CODE_SUCCESS == code && pSelect->timeRange.skey > pSelect->timeRange.ekey) {
pSelect->isEmptyResult = true;
}
return code;
}

16
source/libs/scalar/src/filter.c
View File

@ -1631,12 +1631,7 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
SValueNode *var = (SValueNode *)field->desc;
SDataType *dType = &var->node.resType;
// if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) {
// qDebug("VAL%d => [type:TS][val:[%" PRIi64 "] - [%" PRId64 "]]", i, *(int64_t *)field->data,
// *(((int64_t *)field->data) + 1));
// } else {
qDebug("VAL%d => [type:%d][val:%" PRIx64 "]", i, dType->type, var->datum.i); // TODO
//}
} else if (field->data) {
qDebug("VAL%d => [type:NIL][val:NIL]", i); // TODO
}
@ -1955,20 +1950,9 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
bytes = (len + 1) * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
fi->data = taosMemoryCalloc(1, bytes);
} else {
if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) { // TIME RANGE
/*
fi->data = taosMemoryCalloc(dType->bytes, tDataTypes[type].bytes);
for (int32_t a = 0; a < dType->bytes; ++a) {
int64_t *v = taosArrayGet(var->arr, a);
assignVal((char *)fi->data + a * tDataTypes[type].bytes, (char *)v, 0, type);
}
*/
continue;
} else {
fi->data = taosMemoryCalloc(1, sizeof(int64_t));
}
}
if (dType->type == type) {
assignVal(fi->data, nodesGetValueFromNode(var), dType->bytes, type);

68
source/libs/transport/src/thttp.c
View File

@ -26,6 +26,8 @@
#define HTTP_RECV_BUF_SIZE 1024
static int32_t httpRefMgt = 0;
static int64_t httpRef = -1;
typedef struct SHttpModule {
uv_loop_t* loop;
SAsyncPool* asyncPool;
@ -41,7 +43,6 @@ typedef struct SHttpMsg {
int32_t len;
EHttpCompFlag flag;
int8_t quit;
SHttpModule* http;
} SHttpMsg;
@ -57,7 +58,6 @@ typedef struct SHttpClient {
} SHttpClient;
static TdThreadOnce transHttpInit = PTHREAD_ONCE_INIT;
static SHttpModule* thttp = NULL;
static void transHttpEnvInit();
static void httpHandleReq(SHttpMsg* msg);
@ -280,24 +280,27 @@ static void clientConnCb(uv_connect_t* req, int32_t status) {
}
int32_t httpSendQuit() {
SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef);
if (http == NULL) return 0;
SHttpMsg* msg = taosMemoryCalloc(1, sizeof(SHttpMsg));
msg->quit = 1;
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
httpDestroyMsg(msg);
tError("http-report already released");
return -1;
} else {
msg->http = load;
}
transAsyncSend(load->asyncPool, &(msg->q));
transAsyncSend(http->asyncPool, &(msg->q));
taosReleaseRef(httpRefMgt, httpRef);
return 0;
}
static int32_t taosSendHttpReportImpl(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag) {
SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef);
if (load == NULL) {
tError("http-report already released");
return -1;
}
SHttpMsg* msg = taosMemoryMalloc(sizeof(SHttpMsg));
msg->server = taosStrdup(server);
msg->uri = taosStrdup(uri);
msg->port = port;
@ -307,15 +310,9 @@ static int32_t taosSendHttpReportImpl(const char* server, const char* uri, uint1
msg->flag = flag;
msg->quit = 0;
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
httpDestroyMsg(msg);
tError("http-report already released");
return -1;
}
msg->http = load;
return transAsyncSend(load->asyncPool, &(msg->q));
int ret = transAsyncSend(load->asyncPool, &(msg->q));
taosReleaseRef(httpRefMgt, httpRef);
return ret;
}
static void httpDestroyClientCb(uv_handle_t* handle) {
@ -335,13 +332,19 @@ static void httpWalkCb(uv_handle_t* handle, void* arg) {
return;
}
static void httpHandleQuit(SHttpMsg* msg) {
SHttpModule* http = msg->http;
taosMemoryFree(msg);
SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef);
if (http == NULL) return;
uv_walk(http->loop, httpWalkCb, NULL);
taosReleaseRef(httpRefMgt, httpRef);
}
static void httpHandleReq(SHttpMsg* msg) {
SHttpModule* http = msg->http;
SHttpModule* http = taosAcquireRef(httpRefMgt, httpRef);
if (http == NULL) {
goto END;
}
struct sockaddr_in dest = {0};
if (taosBuildDstAddr(msg->server, msg->port, &dest) < 0) {
@ -391,6 +394,7 @@ static void httpHandleReq(SHttpMsg* msg) {
int ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
if (ret != 0) {
tError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
taosReleaseRef(httpRefMgt, httpRef);
destroyHttpClient(cli);
return;
}
@ -401,21 +405,26 @@ static void httpHandleReq(SHttpMsg* msg) {
cli->port);
destroyHttpClient(cli);
}
taosReleaseRef(httpRefMgt, httpRef);
return;
END:
tError("http-report failed to report, reason: %s, addr: %s:%d", terrstr(), msg->server, msg->port);
httpDestroyMsg(msg);
taosReleaseRef(httpRefMgt, httpRef);
}
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag) {
taosThreadOnce(&transHttpInit, transHttpEnvInit);
return taosSendHttpReportImpl(server, uri, port, pCont, contLen, flag);
}
static void transHttpDestroyHandle(void* handle) { taosMemoryFree(handle); }
static void transHttpEnvInit() {
SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule));
httpRefMgt = taosOpenRef(1, transHttpDestroyHandle);
SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule));
http->loop = taosMemoryMalloc(sizeof(uv_loop_t));
uv_loop_init(http->loop);
@ -433,14 +442,15 @@ static void transHttpEnvInit() {
taosMemoryFree(http);
http = NULL;
}
atomic_store_ptr(&thttp, http);
httpRef = taosAddRef(httpRefMgt, http);
}
void transHttpEnvDestroy() {
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
// remove http
if (httpRef == -1) {
return;
}
SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef);
httpSendQuit();
taosThreadJoin(load->thread, NULL);
@ -448,7 +458,7 @@ void transHttpEnvDestroy() {
transAsyncPoolDestroy(load->asyncPool);
uv_loop_close(load->loop);
taosMemoryFree(load->loop);
taosMemoryFree(load);
atomic_store_ptr(&thttp, NULL);
taosReleaseRef(httpRefMgt, httpRef);
taosRemoveRef(httpRefMgt, httpRef);
}

36
source/libs/transport/src/transCli.c
View File

@ -11,7 +11,9 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "transComm.h"
#include "tutil.h"
typedef struct SConnList {
queue conns;
@ -64,15 +66,13 @@ typedef struct SCliConn {
SCliBatch* pBatch;
int64_t refId;
char* ip;
SDelayTask* task;
// debug and log info
char* ip;
char src[32];
char dst[32];
int64_t refId;
} SCliConn;
typedef struct SCliMsg {
@ -131,6 +131,7 @@ typedef struct {
int32_t threshold;
int64_t interval;
} SFailFastItem;
// conn pool
// add expire timeout and capacity limit
static void* createConnPool(int size);
@ -225,11 +226,11 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
// snprintf may cause performance problem
#define CONN_CONSTRUCT_HASH_KEY(key, ip, port) \
do { \
char* p = key; \
int32_t len = strlen(ip); \
if (p != NULL) memcpy(p, ip, len); \
p[len] = ':'; \
titoa(port, 10, &p[len + 1]); \
char* t = key; \
int16_t len = strlen(ip); \
if (ip != NULL) memcpy(t, ip, len); \
t[len] = ':'; \
titoa(port, 10, &t[len + 1]); \
} while (0)
#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para))
@ -333,13 +334,9 @@ bool cliMaySendCachedMsg(SCliConn* conn) {
if (!transQueueEmpty(&conn->cliMsgs)) {
SCliMsg* pCliMsg = NULL;
CONN_GET_NEXT_SENDMSG(conn);
if (pCliMsg == NULL)
return false;
else {
cliSend(conn);
return true;
}
}
return false;
_RETURN:
return false;
@ -362,6 +359,7 @@ void cliHandleResp(SCliConn* conn) {
int32_t msgLen = transDumpFromBuffer(&conn->readBuf, (char**)&pHead);
if (msgLen <= 0) {
taosMemoryFree(pHead);
tDebug("%s conn %p recv invalid packet ", CONN_GET_INST_LABEL(conn), conn);
return;
}
@ -1730,17 +1728,23 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrd* pThrd = createThrdObj(shandle);
if (pThrd == NULL) {
return NULL;
goto _err;
}
int err = taosThreadCreate(&pThrd->thread, NULL, cliWorkThread, (void*)(pThrd));
if (err == 0) {
if (err != 0) {
goto _err;
} else {
tDebug("success to create tranport-cli thread:%d", i);
}
cli->pThreadObj[i] = pThrd;
}
return cli;
_err:
taosMemoryFree(cli->pThreadObj);
taosMemoryFree(cli);
return NULL;
}
static FORCE_INLINE void destroyUserdata(STransMsg* userdata) {

2
source/util/src/tutil.c
View File

@ -468,7 +468,7 @@ size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize)
c3 = p[s[j + 3]];
if ((c0 | c1 | c2 | c3) != 0) {
size_t count = ((i + 1) >> 2);
size_t count = i * 4;
return (c0 | c1) != 0 ? count - c0 + 1 : count - c2 + 3;
}
}

4
source/util/test/utilTests.cpp
View File

@ -294,6 +294,10 @@ TEST(utilTest, tstrncspn) {
const char* reject5 = "911";
v = tstrncspn(p2, strlen(p2), reject5, 0);
ASSERT_EQ(v, 14);
const char* reject6 = "Kk";
v = tstrncspn(p2, strlen(p2), reject6, 2);
ASSERT_EQ(v, 10);
}
TEST(utilTest, intToHextStr) {

3
tests/parallel_test/cases.task
View File

@ -168,6 +168,7 @@
,,y,script,./test.sh -f tsim/parser/union.sim
,,y,script,./test.sh -f tsim/parser/union_sysinfo.sim
,,y,script,./test.sh -f tsim/parser/where.sim
,,y,script,./test.sh -f tsim/query/tagLikeFilter.sim
,,y,script,./test.sh -f tsim/query/charScalarFunction.sim
,,y,script,./test.sh -f tsim/query/explain.sim
,,y,script,./test.sh -f tsim/query/interval-offset.sim
@ -181,7 +182,7 @@
,,y,script,./test.sh -f tsim/query/groupby.sim
,,y,script,./test.sh -f tsim/query/event.sim
,,y,script,./test.sh -f tsim/query/forceFill.sim
,,n,script,./test.sh -f tsim/query/join.sim
,,y,script,./test.sh -f tsim/query/emptyTsRange.sim
,,y,script,./test.sh -f tsim/qnode/basic1.sim
,,y,script,./test.sh -f tsim/snode/basic1.sim
,,y,script,./test.sh -f tsim/mnode/basic1.sim

20
tests/script/tsim/query/emptyTsRange.sim Normal file
View File

@ -0,0 +1,20 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sql connect
sql drop database if exists db1;
sql create database if not exists db1;
sql use db1;
sql create stable sta (ts timestamp, f1 double, f2 binary(200)) tags(t1 int);
sql create table tba1 using sta tags(1);
sql insert into tba1 values ('2022-04-26 15:15:01', 1.0, "a");
sql insert into tba1 values ('2022-04-26 15:15:02', 2.0, "b");
sql insert into tba1 values ('2022-04-26 15:15:04', 4.0, "b");
sql insert into tba1 values ('2022-04-26 15:15:05', 5.0, "b");
sql select last_row(*) from sta where ts >= 1678901803783 and ts <= 1678901803783 and _c0 <= 1678901803782 interval(10d,8d) fill(linear) order by _wstart desc;
if $rows != 0 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

36
tests/script/tsim/query/tagLikeFilter.sim Normal file
View File

@ -0,0 +1,36 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sql connect
sql drop database if exists db1;
sql create database if not exists db1 vgroups 10;
sql use db1;
sql create stable sta (ts timestamp, f1 double, f2 binary(200)) tags(t1 binary(100));
sql create table tba1 using sta tags('ZQMPvstuzZVzCRjFTQawILuGSqZKSqlJwcBtZMxrAEqBbzChHWVDMiAZJwESzJAf');
sql create table tba2 using sta tags('ieofwehughkreghughuerugu34jf9340aieefjalie28ffj8fj8fafjaekdfjfii');
sql create table tba3 using sta tags('ZQMPvstuzZVzCRjFTQawILuGSqabSqlJwcBtZMxrAEqBbzChHWVDMiAZJwESzJAf');
sql insert into tba1 values ('2022-04-26 15:15:01', 1.0, "a");
sql insert into tba2 values ('2022-04-26 15:15:01', 1.0, "a");
sql insert into tba2 values ('2022-04-26 15:15:02', 1.0, "a");
sql insert into tba3 values ('2022-04-26 15:15:01', 1.0, "a");
sql insert into tba3 values ('2022-04-26 15:15:02', 1.0, "a");
sql insert into tba3 values ('2022-04-26 15:15:03', 1.0, "a");
sql select t1 from sta where t1 like '%ab%';
if $rows != 3 then
return -1
endi
sql select t1 from sta where t1 like '%ax%';
if $rows != 0 then
return -1
endi
sql select t1 from sta where t1 like '%cd%';
if $rows != 0 then
return -1
endi
sql select t1 from sta where t1 like '%ii';
if $rows != 2 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT