提交 c7d35784 编写于 作者: H Hongze Cheng

Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/TD-22694

......@@ -273,49 +273,48 @@ password: taosdata
## Start the TDengine cluster with docker-compose
1. The following docker-compose file starts a TDengine cluster with two replicas, two management nodes, two data nodes, and one arbitrator.
```docker
version: "3"
services:
arbitrator:
image: tdengine/tdengine:$VERSION
command: tarbitrator
td-1:
image: tdengine/tdengine:$VERSION
environment:
TAOS_FQDN: "td-1"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td1:/var/lib/taos/
- taoslog-td1:/var/log/taos/
td-2:
image: tdengine/tdengine:$VERSION
environment:
TAOS_FQDN: "td-2"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td2:/var/lib/taos/
- taoslog-td2:/var/log/taos/
volumes:
taosdata-td1:
taoslog-td1:
taosdata-td2:
taoslog-td2:
```
1. The following docker-compose file starts a TDengine cluster with three nodes.
```yml
version: "3"
services:
td-1:
image: tdengine/tdengine:$VERSION
environment:
TAOS_FQDN: "td-1"
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td1:/var/lib/taos/
- taoslog-td1:/var/log/taos/
td-2:
image: tdengine/tdengine:$VERSION
environment:
TAOS_FQDN: "td-2"
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td2:/var/lib/taos/
- taoslog-td2:/var/log/taos/
td-3:
image: tdengine/tdengine:$VERSION
environment:
TAOS_FQDN: "td-3"
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td3:/var/lib/taos/
- taoslog-td3:/var/log/taos/
volumes:
taosdata-td1:
taoslog-td1:
taosdata-td2:
taoslog-td2:
taosdata-td3:
taoslog-td3:
```
:::note
- The `VERSION` environment variable is used to set the tdengine image tag
- `TAOS_FIRST_EP` must be set on the newly created instance so that it can join the TDengine cluster; if there is a high availability requirement, `TAOS_SECOND_EP` needs to be used at the same time
- `TAOS_REPLICA` is used to set the default number of database replicas. Its value range is [1,3]
We recommend setting it with `TAOS_ARBITRATOR` to use arbitrator in a two-nodes environment.
:::
2. Start the cluster
......@@ -345,17 +344,18 @@ password: taosdata
4. Show dnodes via TDengine CLI
```shell
$ docker-compose exec td-1 taos -s "show dnodes"
taos> show dnodes
id | end_point | vnodes | cores | status | role | create_time | offline reason |
======================================================================================================================================
1 | td-1:6030 | 1 | 8 | ready | any | 2022-01-18 02:47:42.871 | |
2 | td-2:6030 | 0 | 8 | ready | any | 2022-01-18 02:47:43.518 | |
0 | arbitrator:6042 | 0 | 0 | ready | arb | 2022-01-18 02:47:43.633 | - |
Query OK, 3 row(s) in set (0.000811s)
```
```shell
$ docker-compose exec td-1 taos -s "show dnodes"
taos> show dnodes
id | endpoint | vnodes | support_vnodes | status | create_time | note |
======================================================================================================================================
1 | td-1:6030 | 0 | 32 | ready | 2022-08-19 07:57:29.971 | |
2 | td-2:6030 | 0 | 32 | ready | 2022-08-19 07:57:31.415 | |
3 | td-3:6030 | 0 | 32 | ready | 2022-08-19 07:57:31.417 | |
Query OK, 3 rows in database (0.021262s)
```
## taosAdapter
......@@ -373,83 +373,70 @@ password: taosdata
Suppose you want to deploy multiple taosAdapters to improve throughput and provide high availability. In that case, the recommended configuration method uses a reverse proxy such as Nginx to offer a unified access entry. For specific configuration methods, please refer to the official documentation of Nginx. Here is an example:
```docker
version: "3"
networks:
inter:
api:
services:
arbitrator:
image: tdengine/tdengine:$VERSION
command: tarbitrator
networks:
- inter
td-1:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-1"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td1:/var/lib/taos/
- taoslog-td1:/var/log/taos/
td-2:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-2"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td2:/var/lib/taos/
- taoslog-td2:/var/log/taos/
adapter:
image: tdengine/tdengine:$VERSION
command: taosadapter
networks:
- inter
environment:
TAOS_FIRST_EP: "td-1"
TAOS_SECOND_EP: "td-2"
deploy:
replicas: 4
nginx:
image: nginx
depends_on:
- adapter
networks:
- inter
- api
ports:
- 6041:6041
- 6044:6044/udp
command: [
"sh",
"-c",
"while true;
do curl -s http://adapter:6041/-/ping >/dev/null && break;
done;
printf 'server{listen 6041;location /{proxy_pass http://adapter:6041;}}'
> /etc/nginx/conf.d/rest.conf;
printf 'stream{server{listen 6044 udp;proxy_pass adapter:6044;}}'
>> /etc/nginx/nginx.conf;cat /etc/nginx/nginx.conf;
nginx -g 'daemon off;'",
]
volumes:
taosdata-td1:
taoslog-td1:
taosdata-td2:
taoslog-td2:
```
```yml
version: "3"
networks:
inter:
services:
td-1:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-1"
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td1:/var/lib/taos/
- taoslog-td1:/var/log/taos/
td-2:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-2"
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td2:/var/lib/taos/
- taoslog-td2:/var/log/taos/
adapter:
image: tdengine/tdengine:$VERSION
entrypoint: "taosadapter"
networks:
- inter
environment:
TAOS_FIRST_EP: "td-1"
TAOS_SECOND_EP: "td-2"
deploy:
replicas: 4
nginx:
image: nginx
depends_on:
- adapter
networks:
- inter
ports:
- 6041:6041
- 6044:6044/udp
command: [
"sh",
"-c",
"while true;
do curl -s http://adapter:6041/-/ping >/dev/null && break;
done;
printf 'server{listen 6041;location /{proxy_pass http://adapter:6041;}}'
> /etc/nginx/conf.d/rest.conf;
printf 'stream{server{listen 6044 udp;proxy_pass adapter:6044;}}'
>> /etc/nginx/nginx.conf;cat /etc/nginx/nginx.conf;
nginx -g 'daemon off;'",
]
volumes:
taosdata-td1:
taoslog-td1:
taosdata-td2:
taoslog-td2:
```
## Deploy with docker swarm
......
......@@ -309,7 +309,7 @@ services:
TAOS_FIRST_EP: "td-1"
volumes:
- taosdata-td3:/var/lib/taos/
- taoslog-td3:/var/log/taos/
- taoslog-td3:/var/log/taos/
volumes:
taosdata-td1:
taoslog-td1:
......@@ -473,18 +473,18 @@ Creating service taos_adapter
```shell
$ docker stack ps taos
ID NAME IMAGE NODE DESIRED STATE CURRENT STATE ERROR PORTS
7m3sbf532bqp taos_adapter.1 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
pj403n6ofmmh taos_adapter.2 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
rxqfwsyk5q1h taos_adapter.3 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
qj40lpxr40oc taos_adapter.4 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
oe3455ulxpze taos_nginx.1 nginx:latest vm98 Running Running about a minute ago
o0tsg70nrrc6 taos_td-1.1 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
7m3sbf532bqp taos_adapter.1 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
pj403n6ofmmh taos_adapter.2 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
rxqfwsyk5q1h taos_adapter.3 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
qj40lpxr40oc taos_adapter.4 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
oe3455ulxpze taos_nginx.1 nginx:latest vm98 Running Running about a minute ago
o0tsg70nrrc6 taos_td-1.1 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
q5m1oxs589cp taos_td-2.1 tdengine/tdengine:3.0.0.0 vm98 Running Running about a minute ago
$ docker service ls
ID NAME MODE REPLICAS IMAGE PORTS
ozuklorgl8bs taos_adapter replicated 4/4 tdengine/tdengine:3.0.0.0
ozuklorgl8bs taos_adapter replicated 4/4 tdengine/tdengine:3.0.0.0
crmhdjw6vxw0 taos_nginx replicated 1/1 nginx:latest *:6041->6041/tcp, *:6044->6044/udp
o86ngy7csv5n taos_td-1 replicated 1/1 tdengine/tdengine:3.0.0.0
o86ngy7csv5n taos_td-1 replicated 1/1 tdengine/tdengine:3.0.0.0
rma040ny4tb0 taos_td-2 replicated 1/1 tdengine/tdengine:3.0.0.0
```
......@@ -495,11 +495,11 @@ rma040ny4tb0 taos_td-2 replicated 1/1 tdengine/tdengine:3.0.0.0
```shell
$ docker service scale taos_adapter=1
taos_adapter scaled to 1
overall progress: 1 out of 1 tasks
1/1: running [==================================================>]
overall progress: 1 out of 1 tasks
1/1: running [==================================================>]
verify: Service converged
$ docker service ls -f name=taos_adapter
ID NAME MODE REPLICAS IMAGE PORTS
ozuklorgl8bs taos_adapter replicated 1/1 tdengine/tdengine:3.0.0.0
ozuklorgl8bs taos_adapter replicated 1/1 tdengine/tdengine:3.0.0.0
```
......@@ -92,7 +92,7 @@ int main(int argc, char *argv[])
}
// a simple way to parse input parameters
if (argc >= 3) strcpy(db, argv[2]);
if (argc >= 3) strncpy(db, argv[2], sizeof(db) - 1);
if (argc >= 4) points = atoi(argv[3]);
if (argc >= 5) numOfTables = atoi(argv[4]);
......
......@@ -106,7 +106,7 @@ int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
// SRow ================================
int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow);
void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
void tRowDestroy(SRow *pRow);
void tRowSort(SArray *aRowP);
int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, int8_t flag);
......
......@@ -172,8 +172,8 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_SERVER_VERSION, "server-version", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_UPTIME_TIMER, "uptime-timer", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, "lost-consumer-clear", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_STREAM_CHECKPOINT_TIMER, "stream-checkpoint-tmr", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_STREAM_BEGIN_CHECKPOINT, "stream-begin-checkpoint", NULL, NULL)
// TD_DEF_MSG_TYPE(TDMT_MND_STREAM_CHECKPOINT_TIMER, "stream-checkpoint-tmr", NULL, NULL)
// TD_DEF_MSG_TYPE(TDMT_MND_STREAM_BEGIN_CHECKPOINT, "stream-begin-checkpoint", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL)
TD_NEW_MSG_SEG(TDMT_VND_MSG)
......
......@@ -175,20 +175,24 @@ typedef struct {
void streamFreeQitem(SStreamQueueItem* data);
#if 0
bool streamQueueResEmpty(const SStreamQueueRes* pRes);
int64_t streamQueueResSize(const SStreamQueueRes* pRes);
SStreamQueueNode* streamQueueResFront(SStreamQueueRes* pRes);
SStreamQueueNode* streamQueueResPop(SStreamQueueRes* pRes);
void streamQueueResClear(SStreamQueueRes* pRes);
SStreamQueueRes streamQueueBuildRes(SStreamQueueNode* pNode);
#endif
typedef struct {
SStreamQueueNode* pHead;
} SStreamQueue1;
#if 0
bool streamQueueHasTask(const SStreamQueue1* pQueue);
int32_t streamQueuePush(SStreamQueue1* pQueue, SStreamQueueItem* pItem);
SStreamQueueRes streamQueueGetRes(SStreamQueue1* pQueue);
#endif
typedef struct {
STaosQueue* queue;
......@@ -636,7 +640,7 @@ void streamMetaClose(SStreamMeta* streamMeta);
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask);
int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask);
int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, char* msg, int32_t msgLen);
SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
// SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId);
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId);
void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
......
......@@ -119,6 +119,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_APP_IS_STARTING TAOS_DEF_ERROR_CODE(0, 0x0130) //
#define TSDB_CODE_APP_IS_STOPPING TAOS_DEF_ERROR_CODE(0, 0x0131) //
#define TSDB_CODE_IVLD_DATA_FMT TAOS_DEF_ERROR_CODE(0, 0x0132) //
//client
#define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200)
......
此差异已折叠。
version: "3"
networks:
inter:
api:
services:
arbitrator:
image: tdengine/tdengine:$VERSION
command: tarbitrator
networks:
- inter
td-1:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-1"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td1:/var/lib/taos/
- taoslog-td1:/var/log/taos/
td-2:
image: tdengine/tdengine:$VERSION
networks:
- inter
environment:
TAOS_FQDN: "td-2"
TAOS_FIRST_EP: "td-1"
TAOS_NUM_OF_MNODES: "2"
TAOS_REPLICA: "2"
TAOS_ARBITRATOR: arbitrator:6042
volumes:
- taosdata-td2:/var/lib/taos/
- taoslog-td2:/var/log/taos/
adapter:
image: tdengine/tdengine:$VERSION
command: taosadapter
networks:
- inter
environment:
TAOS_FIRST_EP: "td-1"
TOAS_SECOND_EP: "td-2"
deploy:
replicas: 4
update_config:
parallelism: 4
nginx:
image: nginx
depends_on:
- adapter
networks:
- inter
- api
ports:
- 6041:6041
- 6044:6044/udp
command: [
"sh",
"-c",
"while true;
do curl -s http://adapter:6041/-/ping >/dev/null && break;
done;
printf 'server{listen 6041;location /{proxy_pass http://adapter:6041;}}'
> /etc/nginx/conf.d/rest.conf;
printf 'stream{server{listen 6044 udp;proxy_pass adapter:6044;}}'
>> /etc/nginx/nginx.conf;cat /etc/nginx/nginx.conf;
nginx -g 'daemon off;'",
]
volumes:
taosdata-td1:
taoslog-td1:
taosdata-td2:
taoslog-td2:
#!/bin/bash
set -e
#set -x
set -v
set -v
# dockerbuild.sh
# dockerbuild.sh
# -n [version number]
# -p [xxxx]
# -V [stable | beta]
......@@ -28,7 +28,7 @@ do
V)
#echo "verType=$OPTARG"
verType=$(echo $OPTARG)
;;
;;
h)
echo "Usage: `basename $0` -n [version number] "
echo " -p [password for docker hub] "
......@@ -39,8 +39,8 @@ do
a)
#echo "dockerLatest=$OPTARG"
dockerLatest=$(echo $OPTARG)
;;
?) #unknow option
;;
?) #unknow option
echo "unkonw argument"
exit 1
;;
......@@ -60,7 +60,7 @@ if [ "$verType" == "stable" ]; then
elif [ "$verType" == "beta" ];then
verType=beta
tagVal=ver-${version}-beta
dockerinput=TDengine-server-${version}-${verType}-Linux-$cpuType.tar.gz
dockerinput=TDengine-server-${version}-${verType}-Linux-$cpuType.tar.gz
dockerinput_x64=TDengine-server-${version}-${verType}-Linux-amd64.tar.gz
dockerim=tdengine/tdengine-beta
dockeramd64=tdengine/tdengine-amd64-beta
......@@ -73,30 +73,30 @@ fi
username="tdengine"
# generate docker verison
# generate docker version
echo "generate ${dockerim}:${version}"
docker manifest create -a ${dockerim}:${version} ${dockeramd64}:${version} ${dockeraarch64}:${version}
docker manifest inspect ${dockerim}:${version}
docker manifest rm ${dockerim}:${version}
docker manifest create -a ${dockerim}:${version} ${dockeramd64}:${version} ${dockeraarch64}:${version}
docker manifest inspect ${dockerim}:${version}
docker login -u ${username} -p ${passWord}
docker login -u ${username} -p ${passWord}
docker manifest push ${dockerim}:${version}
# generate docker latest
# generate docker latest
echo "generate ${dockerim}:latest "
if [ ${dockerLatest} == 'y' ] ;then
echo "docker manifest create -a ${dockerim}:latest ${dockeramd64}:latest ${dockeraarch64}:latest"
docker manifest create -a ${dockerim}:latest ${dockeramd64}:latest ${dockeraarch64}:latest
docker manifest inspect ${dockerim}:latest
docker manifest rm ${dockerim}:latest
docker manifest inspect ${dockerim}:latest
docker manifest rm ${dockerim}:latest
docker manifest create -a ${dockerim}:latest ${dockeramd64}:latest ${dockeraarch64}:latest
docker manifest inspect ${dockerim}:latest
docker login -u tdengine -p ${passWord} #replace the docker registry username and password
docker manifest push ${dockerim}:latest
docker pull tdengine/tdengine:latest
docker pull tdengine/tdengine:latest
fi
......
......@@ -74,7 +74,7 @@ do
done
# Check_verison()
# Check_version()
# {
# }
......@@ -102,14 +102,14 @@ scriptDir=$(dirname $(readlink -f $0))
communityDir=${scriptDir}/../../../community
DockerfilePath=${communityDir}/packaging/docker/
if [ "$cloudBuild" == "y" ]; then
comunityArchiveDir=/nas/TDengine/v$version/cloud
communityArchiveDir=/nas/TDengine/v$version/cloud
Dockerfile=${communityDir}/packaging/docker/DockerfileCloud
else
comunityArchiveDir=/nas/TDengine/v$version/community
communityArchiveDir=/nas/TDengine/v$version/community
Dockerfile=${communityDir}/packaging/docker/Dockerfile
fi
cd ${scriptDir}
cp -f ${comunityArchiveDir}/${pkgFile} .
cp -f ${communityArchiveDir}/${pkgFile} .
echo "dirName=${dirName}"
......
......@@ -456,12 +456,13 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
(*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS);
int32_t len = blockEncode(pBlock, (*pRsp)->data, SHOW_VARIABLES_RESULT_COLS);
blockDataDestroy(pBlock);
if(len != rspSize - sizeof(SRetrieveTableRsp)){
uError("buildShowVariablesRsp error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len, (uint64_t) (rspSize - sizeof(SRetrieveTableRsp)));
return TSDB_CODE_TSC_INVALID_INPUT;
}
blockDataDestroy(pBlock);
return TSDB_CODE_SUCCESS;
}
......
......@@ -139,7 +139,10 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
nkv += tPutI16v(NULL, -pTColumn->colId);
nIdx++;
} else {
ASSERT(0);
if (ASSERTS(0, "invalid input")) {
code = TSDB_CODE_INVALID_PARA;
goto _exit;
}
}
pTColumn = (++iTColumn < pTSchema->numOfCols) ? pTSchema->columns + iTColumn : NULL;
......@@ -176,8 +179,10 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
ntp = sizeof(SRow) + BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntp;
break;
default:
ASSERT(0);
break;
if (ASSERTS(0, "impossible")) {
code = TSDB_CODE_INVALID_PARA;
goto _exit;
}
}
if (maxIdx <= UINT8_MAX) {
nkv = sizeof(SRow) + sizeof(SKVIdx) + nIdx + nkv;
......@@ -306,8 +311,10 @@ int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
pv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
if (ASSERTS(0, "impossible")) {
code = TSDB_CODE_INVALID_PARA;
goto _exit;
}
}
if (pb) {
......@@ -370,7 +377,7 @@ _exit:
return code;
}
void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
int32_t tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
ASSERT(iCol < pTSchema->numOfCols);
ASSERT(pRow->sver == pTSchema->version);
......@@ -381,17 +388,17 @@ void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
pColVal->type = pTColumn->type;
pColVal->flag = CV_FLAG_VALUE;
memcpy(&pColVal->value.val, &pRow->ts, sizeof(TSKEY));
return;
return 0;
}
if (pRow->flag == HAS_NONE) {
*pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
return;
return 0;
}
if (pRow->flag == HAS_NULL) {
*pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
return;
return 0;
}
if (pRow->flag >> 4) { // KV Row
......@@ -440,7 +447,7 @@ void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
memcpy(&pColVal->value.val, pData, pTColumn->bytes);
}
}
return;
return 0;
} else if (TABS(cid) < pTColumn->colId) {
lidx = mid + 1;
} else {
......@@ -492,16 +499,16 @@ void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
pv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
ASSERTS(0, "invalid row format");
return TSDB_CODE_IVLD_DATA_FMT;
}
if (bv == BIT_FLG_NONE) {
*pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
return;
return 0;
} else if (bv == BIT_FLG_NULL) {
*pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
return;
return 0;
}
pColVal->cid = pTColumn->colId;
......@@ -520,6 +527,8 @@ void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
}
}
}
return 0;
}
void tRowDestroy(SRow *pRow) {
......@@ -710,7 +719,6 @@ int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter) {
_exit:
if (code) {
*ppIter = NULL;
if (pIter) taosMemoryFree(pIter);
} else {
*ppIter = pIter;
}
......@@ -929,8 +937,8 @@ static int32_t tRowTupleUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *
pv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
break;
ASSERTS(0, "Invalid row flag");
return TSDB_CODE_IVLD_DATA_FMT;
}
while (pColData) {
......@@ -954,8 +962,8 @@ static int32_t tRowTupleUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *
bv = GET_BIT2(pb, iTColumn - 1);
break;
default:
ASSERT(0);
break;
ASSERTS(0, "Invalid row flag");
return TSDB_CODE_IVLD_DATA_FMT;
}
if (bv == BIT_FLG_NONE) {
......@@ -1045,7 +1053,8 @@ static int32_t tRowKVUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aCo
} else if (pRow->flag & KV_FLG_BIG) {
pData = pv + ((uint32_t *)pKVIdx->idx)[iCol];
} else {
ASSERT(0);
ASSERTS(0, "Invalid KV row format");
return TSDB_CODE_IVLD_DATA_FMT;
}
int16_t cid;
......@@ -1579,7 +1588,7 @@ static FORCE_INLINE int32_t tColDataPutValue(SColData *pColData, uint8_t *pData,
int32_t code = 0;
if (IS_VAR_DATA_TYPE(pColData->type)) {
code = tRealloc((uint8_t **)(&pColData->aOffset), (pColData->nVal + 1) << 2);
code = tRealloc((uint8_t **)(&pColData->aOffset), ((int64_t)(pColData->nVal + 1)) << 2);
if (code) goto _exit;
pColData->aOffset[pColData->nVal] = pColData->nData;
......@@ -2347,35 +2356,25 @@ void tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal) {
}
uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal) {
uint8_t v;
switch (pColData->flag) {
case HAS_NONE:
v = 0;
break;
return 0;
case HAS_NULL:
v = 1;
break;
return 1;
case (HAS_NULL | HAS_NONE):
v = GET_BIT1(pColData->pBitMap, iVal);
break;
return GET_BIT1(pColData->pBitMap, iVal);
case HAS_VALUE:
v = 2;
break;
return 2;
case (HAS_VALUE | HAS_NONE):
v = GET_BIT1(pColData->pBitMap, iVal);
if (v) v = 2;
break;
return (GET_BIT1(pColData->pBitMap, iVal)) ? 2 : 0;
case (HAS_VALUE | HAS_NULL):
v = GET_BIT1(pColData->pBitMap, iVal) + 1;
break;
return GET_BIT1(pColData->pBitMap, iVal) + 1;
case (HAS_VALUE | HAS_NULL | HAS_NONE):
v = GET_BIT2(pColData->pBitMap, iVal);
break;
return GET_BIT2(pColData->pBitMap, iVal);
default:
ASSERT(0);
break;
ASSERTS(0, "not possible");
return 0;
}
return v;
}
int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg) {
......
......@@ -638,7 +638,7 @@ static int32_t mndSetUpdateIdxStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStb
}
int32_t mndAddIndexImpl(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, SIdxObj *pIdx) {
// impl later
int32_t code = 0;
int32_t code = -1;
SStbObj newStb = {0};
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb-index");
if (pTrans == NULL) goto _OVER;
......@@ -670,6 +670,7 @@ _OVER:
}
static int32_t mndAddIndex(SMnode *pMnode, SRpcMsg *pReq, SCreateTagIndexReq *req, SDbObj *pDb, SStbObj *pStb) {
int32_t code = -1;
SIdxObj idxObj = {0};
memcpy(idxObj.name, req->idxName, TSDB_TABLE_FNAME_LEN);
memcpy(idxObj.stb, pStb->name, TSDB_TABLE_FNAME_LEN);
......@@ -681,21 +682,6 @@ static int32_t mndAddIndex(SMnode *pMnode, SRpcMsg *pReq, SCreateTagIndexReq *re
idxObj.stbUid = pStb->uid;
idxObj.dbUid = pStb->dbUid;
int32_t code = -1;
// SField *pField0 = NULL;
// SStbObj stbObj = {0};
// SStbObj *pNew = &stbObj;
// taosRLockLatch(&pOld->lock);
// memcpy(&stbObj, pOld, sizeof(SStbObj));
// taosRUnLockLatch(&pOld->lock);
// stbObj.pColumns = NULL;
// stbObj.pTags = NULL;
// stbObj.updateTime = taosGetTimestampMs();
// stbObj.lock = 0;
int32_t tag = mndFindSuperTableTagId(pStb, req->colName);
if (tag < 0) {
terrno = TSDB_CODE_MND_TAG_NOT_EXIST;
......
......@@ -133,6 +133,7 @@ static void mndCalMqRebalance(SMnode *pMnode) {
}
}
#if 0
static void mndStreamCheckpointTick(SMnode *pMnode, int64_t sec) {
int32_t contLen = 0;
void *pReq = mndBuildCheckpointTickMsg(&contLen, sec);
......@@ -145,6 +146,7 @@ static void mndStreamCheckpointTick(SMnode *pMnode, int64_t sec) {
tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
}
}
#endif
static void mndPullupTelem(SMnode *pMnode) {
mTrace("pullup telem msg");
......
......@@ -39,7 +39,7 @@ static int32_t mndStreamActionUpdate(SSdb *pSdb, SStreamObj *pStream, SStreamObj
static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq);
static int32_t mndProcessDropStreamReq(SRpcMsg *pReq);
static int32_t mndProcessStreamCheckpointTmr(SRpcMsg *pReq);
static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq);
// static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq);
/*static int32_t mndProcessRecoverStreamReq(SRpcMsg *pReq);*/
static int32_t mndProcessStreamMetaReq(SRpcMsg *pReq);
static int32_t mndGetStreamMeta(SRpcMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
......@@ -66,8 +66,8 @@ int32_t mndInitStream(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_STREAM_TASK_DEPLOY_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_STREAM_TASK_DROP_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_STREAM_CHECKPOINT_TIMER, mndProcessStreamCheckpointTmr);
mndSetMsgHandle(pMnode, TDMT_MND_STREAM_BEGIN_CHECKPOINT, mndProcessStreamDoCheckpoint);
// mndSetMsgHandle(pMnode, TDMT_MND_STREAM_CHECKPOINT_TIMER, mndProcessStreamCheckpointTmr);
// mndSetMsgHandle(pMnode, TDMT_MND_STREAM_BEGIN_CHECKPOINT, mndProcessStreamDoCheckpoint);
mndSetMsgHandle(pMnode, TDMT_STREAM_TASK_REPORT_CHECKPOINT, mndTransProcessRsp);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STREAMS, mndRetrieveStream);
......@@ -778,6 +778,9 @@ _OVER:
tFreeStreamObj(&streamObj);
return code;
}
#if 0
static int32_t mndProcessStreamCheckpointTmr(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
......@@ -942,6 +945,8 @@ static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq) {
return 0;
}
#endif
static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
SStreamObj *pStream = NULL;
......
......@@ -328,10 +328,6 @@ static int32_t tsdbSnapCmprTombData(STsdbSnapReader* pReader, uint8_t** ppData)
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pReader->pTsdb->pVnode), __func__, lino, tstrerror(code));
if (pData) {
taosMemoryFree(pData);
pData = NULL;
}
}
*ppData = pData;
return code;
......@@ -404,7 +400,7 @@ static int32_t tsdbSnapReadTombData(STsdbSnapReader* pReader, uint8_t** ppData)
}
while (pDelInfo && pDelInfo->suid == pReader->tbid.suid && pDelInfo->uid == pReader->tbid.uid) {
if (taosArrayPush(pReader->aDelData, &pDelInfo->delData) < 0) {
if (taosArrayPush(pReader->aDelData, &pDelInfo->delData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
......@@ -1252,7 +1248,7 @@ static int32_t tsdbSnapWriteDelTableData(STsdbSnapWriter* pWriter, TABLEID* pId,
SDelData delData;
n += tGetDelData(pData + n, &delData);
if (taosArrayPush(pWriter->aDelData, &delData) < 0) {
if (taosArrayPush(pWriter->aDelData, &delData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
......@@ -1420,6 +1416,7 @@ _exit:
tBlockDataDestroy(&pWriter->bData);
tBlockDataDestroy(&pWriter->inData);
tsdbFSDestroy(&pWriter->fs);
taosMemoryFree(pWriter);
pWriter = NULL;
}
} else {
......
......@@ -116,12 +116,7 @@ int32_t tMapDataToArray(SMapData *pMapData, int32_t itemSize, int32_t (*tGetItem
}
_exit:
if (code) {
*ppArray = NULL;
if (pArray) taosArrayDestroy(pArray);
} else {
*ppArray = pArray;
}
*ppArray = pArray;
return code;
}
......@@ -1233,14 +1228,22 @@ int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema,
int32_t iColumn = 1;
STColumn *pTColumn = &pTSchema->columns[iColumn];
for (int32_t iCid = 0; iCid < nCid; iCid++) {
ASSERT(pTColumn);
if (ASSERTS(pTColumn != NULL, "invalid input param")) {
code = TSDB_CODE_INVALID_PARA;
goto _exit;
}
while (pTColumn->colId < aCid[iCid]) {
iColumn++;
ASSERT(iColumn < pTSchema->numOfCols);
pTColumn = &pTSchema->columns[iColumn];
}
ASSERT(pTColumn->colId == aCid[iCid]);
if (ASSERTS(pTColumn->colId == aCid[iCid], "invalid input param")) {
code = TSDB_CODE_INVALID_PARA;
goto _exit;
}
tColDataInit(&pBlockData->aColData[iCid], pTColumn->colId, pTColumn->type,
(pTColumn->flags & COL_SMA_ON) ? 1 : 0);
......
......@@ -31,7 +31,9 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq2 *pMsg, SSubmitRsp2
int32_t affectedrows = 0;
int32_t numOfRows = 0;
ASSERT(pTsdb->mem != NULL);
if (ASSERTS(pTsdb->mem != NULL, "vgId:%d, mem is NULL", TD_VID(pTsdb->pVnode))) {
return -1;
}
if (pMsg) {
arrSize = taosArrayGetSize(pMsg->aSubmitTbData);
......
......@@ -36,8 +36,8 @@ static int32_t vnodeCompactTask(void *param) {
vnodeCommitInfo(dir);
_exit:
taosMemoryFree(pInfo);
tsem_post(&pInfo->pVnode->canCommit);
taosMemoryFree(pInfo);
return code;
}
static int32_t vnodePrepareCompact(SVnode *pVnode, SCompactInfo *pInfo) {
......@@ -59,9 +59,17 @@ static int32_t vnodePrepareCompact(SVnode *pVnode, SCompactInfo *pInfo) {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
vnodeLoadInfo(dir, &info);
if (vnodeLoadInfo(dir, &info) < 0) {
code = terrno;
goto _exit;
}
info.state.commitID = pInfo->commitID;
vnodeSaveInfo(dir, &info);
if (vnodeSaveInfo(dir, &info) < 0) {
code = terrno;
goto _exit;
}
_exit:
if (code) {
......
......@@ -48,7 +48,7 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
info.state.applied = -1;
info.state.commitID = 0;
vInfo("vgId:%d, save config while create", pCfg->vgId);
vInfo("vgId:%d, save config while create", info.config.vgId);
if (vnodeSaveInfo(dir, &info) < 0 || vnodeCommitInfo(dir) < 0) {
vError("vgId:%d, failed to save vnode config since %s", pCfg ? pCfg->vgId : 0, tstrerror(terrno));
return -1;
......@@ -124,7 +124,7 @@ int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t sr
while (1) {
const STfsFile *tsdbFile = tfsReaddir(tsdbDir);
if (tsdbFile == NULL) break;
if (tsdbFile->rname == NULL) continue;
if (tsdbFile->rname[0] == '\0') continue;
tstrncpy(oldRname, tsdbFile->rname, TSDB_FILENAME_LEN);
char *tsdbFilePrefixPos = strstr(oldRname, tsdbFilePrefix);
......
......@@ -141,7 +141,10 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
*(int64_t *)(pCoder->data + pCoder->pos) = uid;
pCoder->pos += sizeof(int64_t);
} else {
tDecodeI64(pCoder, &submitTbData.uid);
if (tDecodeI64(pCoder, &submitTbData.uid) < 0) {
code = TSDB_CODE_INVALID_MSG;
TSDB_CHECK_CODE(code, lino, _exit);
}
}
if (tDecodeI32v(pCoder, &submitTbData.sver) < 0) {
......@@ -168,6 +171,11 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
SColData colData = {0};
pCoder->pos += tGetColData(pCoder->data + pCoder->pos, &colData);
if (colData.flag != HAS_VALUE) {
code = TSDB_CODE_INVALID_MSG;
goto _exit;
}
for (int32_t iRow = 0; iRow < colData.nVal; iRow++) {
if (((TSKEY *)colData.pData)[iRow] < minKey || ((TSKEY *)colData.pData)[iRow] > maxKey) {
code = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
......
......@@ -493,11 +493,9 @@ int32_t ctgCopyTbMeta(SCatalog *pCtg, SCtgTbMetaCtx *ctx, SCtgDBCache **pDb, SCt
//ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
if (tbCache) {
CTG_UNLOCK(CTG_READ, &tbCache->metaLock);
taosHashRelease(dbCache->tbCache, tbCache);
*pTb = NULL;
}
CTG_UNLOCK(CTG_READ, &tbCache->metaLock);
taosHashRelease(dbCache->tbCache, tbCache);
*pTb = NULL;
ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", ctx->pName->tname,
ctx->tbInfo.tbType, dbFName);
......@@ -1554,8 +1552,8 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
SCtgTbCache cache = {0};
cache.pMeta = meta;
if (taosHashPut(dbCache->tbCache, tbName, strlen(tbName), &cache, sizeof(SCtgTbCache)) != 0) {
taosMemoryFree(meta);
ctgError("taosHashPut new tbCache failed, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
taosMemoryFree(meta);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
......
......@@ -1470,8 +1470,8 @@ static int32_t translateInterpFunc(STranslateContext* pCxt, SFunctionNode* pFunc
SNode* pTable = pSelect->pFromTable;
if ((NULL != pTable && (QUERY_NODE_REAL_TABLE != nodeType(pTable) ||
(TSDB_CHILD_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType &&
TSDB_NORMAL_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType)))) {
(TSDB_CHILD_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType &&
TSDB_NORMAL_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType)))) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE,
"%s is only supported in single table query", pFunc->functionName);
}
......@@ -5953,6 +5953,7 @@ static int32_t adjustOrderOfProjections(STranslateContext* pCxt, SNodeList* pCol
}
int32_t code = TSDB_CODE_SUCCESS;
bool hasPrimaryKey = false;
SNode* pCol = NULL;
SNode* pProj = NULL;
FORBOTH(pCol, pCols, pProj, *pProjections) {
......@@ -5966,6 +5967,14 @@ static int32_t adjustOrderOfProjections(STranslateContext* pCxt, SNodeList* pCol
if (TSDB_CODE_SUCCESS != code) {
break;
}
if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
hasPrimaryKey = true;
}
}
if (TSDB_CODE_SUCCESS == code && !hasPrimaryKey) {
code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMNS_NUM,
"primary timestamp column can not be null");
}
SNodeList* pNewProjections = NULL;
......@@ -6008,7 +6017,15 @@ static int32_t adjustProjectionsForExistTable(STranslateContext* pCxt, SCreateSt
return adjustOrderOfProjections(pCxt, pStmt->pCols, pMeta, &pSelect->pProjectionList, pReq);
}
static bool isGroupIdTagStream(const STableMeta* pMeta, SNodeList* pTags) {
return (NULL == pTags && 1 == pMeta->tableInfo.numOfTags && TSDB_DATA_TYPE_UBIGINT == getTableTagSchema(pMeta)->type);
}
static int32_t adjustDataTypeOfTags(STranslateContext* pCxt, const STableMeta* pMeta, SNodeList* pTags) {
if (isGroupIdTagStream(pMeta, pTags)) {
return TSDB_CODE_SUCCESS;
}
if (getNumOfTags(pMeta) != LIST_LENGTH(pTags)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMNS_NUM, "Illegal number of tags");
}
......
......@@ -847,6 +847,9 @@ _return:
qwBuildAndSendFetchRsp(qwMsg->msgType + 1, &qwMsg->connInfo, rsp, dataLen, code);
QW_TASK_DLOG("%s send, handle:%p, code:%x - %s, dataLen:%d", TMSG_INFO(qwMsg->msgType + 1),
qwMsg->connInfo.handle, code, tstrerror(code), dataLen);
} else {
qwFreeFetchRsp(rsp);
rsp = NULL;
}
}
......@@ -1217,7 +1220,7 @@ void qWorkerStopAllTasks(void *qWorkerMgmt) {
QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED);
QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);
} else {
qwDropTask(QW_FPARAMS());
(void)qwDropTask(QW_FPARAMS());
}
QW_UNLOCK(QW_WRITE, &ctx->lock);
......
......@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#if 0
#include "streamInc.h"
int32_t tEncodeSStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {
......@@ -192,3 +193,4 @@ int32_t streamProcessCheckpointRsp(SStreamMeta* pMeta, SStreamTask* pTask, SStre
// set status normal
return 0;
}
#endif
......@@ -26,7 +26,7 @@ int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock
ASSERT(pReq->blockNum == taosArrayGetSize(pReq->dataLen));
for (int32_t i = 0; i < blockNum; i++) {
SRetrieveTableRsp* pRetrieve = taosArrayGetP(pReq->data, i);
SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*) taosArrayGetP(pReq->data, i);
SSDataBlock* pDataBlock = taosArrayGet(pArray, i);
blockDecode(pDataBlock, pRetrieve->data);
// TODO: refactor
......
......@@ -114,12 +114,14 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
int32_t batchCnt = 0;
while (1) {
if (atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING) {
taosArrayDestroy(pRes);
return 0;
}
SSDataBlock* output = NULL;
uint64_t ts = 0;
if (qExecTask(exec, &output, &ts) < 0) {
taosArrayDestroy(pRes);
return -1;
}
if (output == NULL) {
......
......@@ -171,6 +171,7 @@ int32_t streamMetaAddTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask) {
}
#endif
#if 0
SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId) {
SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t));
if (ppTask) {
......@@ -180,6 +181,7 @@ SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId) {
return NULL;
}
}
#endif
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId) {
taosRLockLatch(&pMeta->lock);
......
......@@ -46,6 +46,7 @@ void streamQueueClose(SStreamQueue* queue) {
taosMemoryFree(queue);
}
#if 0
bool streamQueueResEmpty(const SStreamQueueRes* pRes) {
//
return true;
......@@ -101,3 +102,4 @@ SStreamQueueRes streamQueueGetRes(SStreamQueue1* pQueue) {
if (pNode) return streamQueueBuildRes(pNode);
return (SStreamQueueRes){0};
}
#endif
......@@ -97,6 +97,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TIMEOUT_ERROR, "Operation timeout")
TAOS_DEFINE_ERROR(TSDB_CODE_APP_IS_STARTING, "Database is starting up")
TAOS_DEFINE_ERROR(TSDB_CODE_APP_IS_STOPPING, "Database is closing down")
TAOS_DEFINE_ERROR(TSDB_CODE_IVLD_DATA_FMT, "Invalid data format")
//client
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_OPERATION, "Invalid operation")
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册