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

Merge branch 'main' into doc/contrib

This commit is contained in:
Shengliang Guan 2024-11-13 13:38:36 +08:00
parent aebb2f658f 5cdeadee38
commit 3c640c2255
20 changed files with 3572 additions and 2215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
Jenkinsfile2
View File

@ -1,9 +1,11 @@
import hudson.model.Result
import hudson.model.*;
import jenkins.model.CauseOfInterruption
docs_only=0
node {
}
file_zh_changed = ''
file_en_changed = ''
file_no_doc_changed = ''
def abortPreviousBuilds() {
def currentJobName = env.JOB_NAME
def currentBuildNumber = env.BUILD_NUMBER.toInteger()
@ -40,39 +42,94 @@ def check_docs() {
cd ${WKC}
git reset --hard
git clean -f
rm -rf examples/rust/
git remote prune origin
git fetch
'''
script {
sh '''
cd ${WKC}
git checkout ''' + env.CHANGE_TARGET + '''
'''
}
sh '''
cd ${WKC}
git remote prune origin
git pull >/dev/null
git fetch origin +refs/pull/${CHANGE_ID}/merge
git checkout -qf FETCH_HEAD
'''
def file_changed = sh (
file_zh_changed = sh (
script: '''
cd ${WKC}
git --no-pager diff --name-only FETCH_HEAD `git merge-base FETCH_HEAD ${CHANGE_TARGET}`|grep -v "^docs/en/"|grep -v "^docs/zh/" || :
git --no-pager diff --name-only FETCH_HEAD `git merge-base FETCH_HEAD ${CHANGE_TARGET}`|grep "^docs/zh/" || :
''',
returnStdout: true
)
file_en_changed = sh (
script: '''
cd ${WKC}
git --no-pager diff --name-only FETCH_HEAD `git merge-base FETCH_HEAD ${CHANGE_TARGET}`|grep "^docs/en/" || :
''',
returnStdout: true
)
file_no_doc_changed = sh (
script: '''
cd ${WKC}
git --no-pager diff --name-only FETCH_HEAD `git merge-base FETCH_HEAD ${CHANGE_TARGET}`|grep -v "^docs/en/"|grep -v "^docs/zh/"|grep -v "*.md" || :
''',
returnStdout: true
).trim()
if (file_changed == '') {
echo "docs PR"
docs_only=1
} else {
echo file_changed
}
env.FILE_CHANGED = file_changed
echo "file_zh_changed: ${file_zh_changed}"
echo "file_en_changed: ${file_en_changed}"
echo "file_no_doc_changed: ${file_no_doc_changed}"
}
}
def build_pre_docs(){
if (env.CHANGE_URL =~ /\/TDengine\//) {
sh '''
hostname
date
env
'''
sh '''
cd ${DOC_WKC}/${td_repo}
git reset --hard
git clean -f
git remote prune origin
git fetch
git checkout ''' + env.CHANGE_TARGET + '''
git pull >/dev/null
git fetch origin +refs/pull/${CHANGE_ID}/merge
git checkout -qf FETCH_HEAD
'''
sh '''
cd ${DOC_WKC}/${tools_repo}
git reset --hard
git clean -f
git fetch
git remote prune origin
git checkout ''' + env.CHANGE_TARGET + '''
git pull >/dev/null
'''
}
}
def build_zh_docs(){
sh '''
cd ${DOC_WKC}/${zh_doc_repo}
# git pull
yarn ass local
yarn build
'''
}
def build_en_docs(){
sh '''
cd ${DOC_WKC}/${en_doc_repo}
# git pull
yarn ass local
yarn build
'''
}
def pre_test(){
sh '''
hostname
@ -153,6 +210,7 @@ def pre_test(){
'''
return 1
}
def pre_test_build_mac() {
sh '''
hostname
@ -173,6 +231,7 @@ def pre_test_build_mac() {
date
'''
}
def pre_test_win(){
bat '''
hostname
@ -284,6 +343,7 @@ def pre_test_win(){
git log -5
'''
}
def pre_test_build_win() {
bat '''
echo "building ..."
@ -313,6 +373,7 @@ def pre_test_build_win() {
'''
return 1
}
def run_win_ctest() {
bat '''
echo "windows ctest ..."
@ -322,6 +383,7 @@ def run_win_ctest() {
time /t
'''
}
def run_win_test() {
bat '''
echo "windows test ..."
@ -344,28 +406,67 @@ pipeline {
WK = '/var/lib/jenkins/workspace/TDinternal'
WKC = '/var/lib/jenkins/workspace/TDinternal/community'
WKPY = '/var/lib/jenkins/workspace/taos-connector-python'
DOC_WKC = '/root/doc_ci_work'
td_repo = 'TDengine'
zh_doc_repo = 'docs.taosdata.com'
en_doc_repo = 'docs.tdengine.com'
tools_repo = 'taos-tools'
}
stages {
stage('check') {
when {
allOf {
not { expression { env.CHANGE_BRANCH =~ /docs\// }}
}
}
parallel {
stage('check docs') {
stage ('check doc file changed') {
agent{label " slave1_47 || slave1_48 || slave1_49 || slave1_50 || slave1_52 || slave1_59 || slave1_63 || worker03 || slave215 || slave217 || slave219 || Mac_catalina "}
steps {
check_docs()
}
}
stage ('pre for build docs') {
when {
beforeAgent true
expression { env.CHANGE_BRANCH =~ /(?i)doc.*/ || file_zh_changed != '' || file_en_changed != '' }
}
agent{label "doc_build_0_30"}
steps {
build_pre_docs()
}
}
stage('build Docs') {
when {
beforeAgent true
expression { env.CHANGE_BRANCH =~ /(?i)doc.*/ || file_zh_changed != '' || file_en_changed != '' }
}
parallel {
stage('build zh docs') {
agent{label "doc_build_0_30"}
when {
expression { file_zh_changed != '' }
}
steps {
build_zh_docs()
}
}
stage('build en docs') {
agent{label "doc_build_0_30"}
when {
expression { file_en_changed != '' }
}
steps {
build_en_docs()
}
}
}
post {
unsuccessful {
error('build docs stage failed, terminating pipeline.')
}
}
}
stage('run test') {
when {
allOf {
not { expression { env.CHANGE_BRANCH =~ /docs\// }}
expression { docs_only == 0 }
not { expression { file_no_doc_changed == '' }}
}
}
parallel {
@ -420,7 +521,7 @@ pipeline {
script {
sh '''
mkdir -p ${WKDIR}/tmp/${BRANCH_NAME}_${BUILD_ID}
echo "''' + env.FILE_CHANGED + '''" > ${WKDIR}/tmp/${BRANCH_NAME}_${BUILD_ID}/docs_changed.txt
echo "''' + file_no_doc_changed + '''" > ${WKDIR}/tmp/${BRANCH_NAME}_${BUILD_ID}/docs_changed.txt
'''
sh '''
cd ${WKC}/tests/parallel_test

2
docs/en/14-reference/03-taos-sql/10-function.md
View File

@ -1149,7 +1149,7 @@ TOP(expr, k)
UNIQUE(expr)
```
**Description**: The values that occur the first time in the specified column. The effect is similar to `distinct` keyword. For a table with composite primary key, only the data with the smallest primary key value is returned.
**Description**: Return the unique values of this column. The effect is similar to `distinct` keyword. Return the row with the earliest timestamp for duplicate data. For a table with composite primary key, only the data with the smallest primary key value is returned.
**Return value type**:Same as the data type of the column being operated upon

4
docs/zh/08-operation/06-gui.md
View File

@ -165,6 +165,10 @@ toc_max_heading_level: 4
第一步 填写添加新主题需要的信息，点击“创建”按钮；
![topic-03-addTopicWizard.jpeg](./pic/topic-03-addTopicWizard.jpeg "添加新主题 Wizard 页面")
如上图，您可以选择是否 “同步 meta”。如果同步 meta 信息，则可以订阅到 meta 信息，比如增加或者删除超级表。
您需要根据使用场景来选择是否开启，如果您引用 taos 连接器编写业务代码订阅 topic，则不能开启“同步 meta”，只能订阅数据；如果您创建 topic 在 explorer 配置同步任务使用，则可以开启“同步 meta”。
第二步 页面出现以下记录，则证明创建成功。
![topic-05-addTopicSucc1.jpeg](./pic/topic-05-addTopicSucc1.jpeg "查看已创建的流计算")

BIN
docs/zh/08-operation/pic/topic-03-addTopicWizard.jpeg
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 275 KiB

After

Width:  |  Height:  |  Size: 83 KiB

2
docs/zh/14-reference/03-taos-sql/10-function.md
View File

@ -1988,7 +1988,7 @@ TOP(expr, k)
UNIQUE(expr)
```
**功能说明**：返回该列数据首次出现的值。该函数功能与 distinct 相似。对于存在复合主键的表的查询，若最小时间戳的数据有多条，则只有对应的复合主键最小的数据被返回。
**功能说明**：返回该列数据去重后的值。该函数功能与 distinct 相似。对于相同的数据，返回时间戳最小的一条，对于存在复合主键的表的查询，若最小时间戳的数据有多条，则只有对应的复合主键最小的数据被返回。
**返回数据类型**：同应用的字段。

22
include/util/tutil.h
View File

@ -120,6 +120,18 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
}
}
/*
* LIKELY and UNLIKELY macros for branch predication hints. Use them judiciously
* only in very hot code paths. Misuse or abuse can lead to performance degradation.
*/
#if __GNUC__ >= 3
#define LIKELY(x) __builtin_expect((x) != 0, 1)
#define UNLIKELY(x) __builtin_expect((x) != 0, 0)
#else
#define LIKELY(x) ((x) != 0)
#define UNLIKELY(x) ((x) != 0)
#endif
#define TAOS_CHECK_ERRNO(CODE) \
do { \
terrno = (CODE); \
@ -131,7 +143,7 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
#define TSDB_CHECK_CODE(CODE, LINO, LABEL) \
do { \
if (TSDB_CODE_SUCCESS != (CODE)) { \
if (UNLIKELY(TSDB_CODE_SUCCESS != (CODE))) { \
LINO = __LINE__; \
goto LABEL; \
} \
@ -139,15 +151,17 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
#define QUERY_CHECK_CODE TSDB_CHECK_CODE
#define QUERY_CHECK_CONDITION(condition, CODE, LINO, LABEL, ERRNO) \
if (!condition) { \
#define TSDB_CHECK_CONDITION(condition, CODE, LINO, LABEL, ERRNO) \
if (UNLIKELY(!(condition))) { \
(CODE) = (ERRNO); \
(LINO) = __LINE__; \
goto LABEL; \
}
#define QUERY_CHECK_CONDITION TSDB_CHECK_CONDITION
#define TSDB_CHECK_NULL(ptr, CODE, LINO, LABEL, ERRNO) \
if ((ptr) == NULL) { \
if (UNLIKELY((ptr) == NULL)) { \
(CODE) = (ERRNO); \
(LINO) = __LINE__; \
goto LABEL; \

8
packaging/smokeTest/test_server.py
View File

@ -61,7 +61,7 @@ def setup_module(get_config):
else:
cmd = "mkdir -p ../../debug/build/bin/"
subprocess.getoutput(cmd)
if config["system"] == "Linux": # add tmq_sim
if config["system"] == "Linux" or config["system"] == "Darwin" : # add tmq_sim
cmd = "cp -rf ../../../debug/build/bin/tmq_sim ../../debug/build/bin/."
subprocess.getoutput(cmd)
if config["system"] == "Darwin":
@ -140,9 +140,11 @@ class TestServer:
if line:
print(line.strip())
if "succeed to write dnode" in line:
time.sleep(15)
time.sleep(5)
# 发送终止信号
os.kill(process.pid, signal.SIGTERM)
os.kill(process.pid, signal.SIGKILL)
# Waiting for the process to be completely killed
time.sleep(5)
break
@pytest.mark.all

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

@ -172,7 +172,7 @@ void tsdbReleaseDataBlock2(STsdbReader *pReader);
int32_t tsdbRetrieveDataBlock2(STsdbReader *pReader, SSDataBlock **pBlock, SArray *pIdList);
int32_t tsdbReaderReset2(STsdbReader *pReader, SQueryTableDataCond *pCond);
int32_t tsdbGetFileBlocksDistInfo2(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
int64_t tsdbGetNumOfRowsInMemTable2(STsdbReader *pHandle);
int64_t tsdbGetNumOfRowsInMemTable2(STsdbReader *pHandle, uint32_t *rows);
void *tsdbGetIdx2(SMeta *pMeta);
void *tsdbGetIvtIdx2(SMeta *pMeta);
uint64_t tsdbGetReaderMaxVersion2(STsdbReader *pReader);

316
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -25,82 +25,109 @@
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
static int32_t setFirstLastResColToNull(SColumnInfoData* pCol, int32_t row) {
char* buf = taosMemoryCalloc(1, pCol->info.bytes);
if (buf == NULL) {
return terrno;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
char* buf = NULL;
SFirstLastRes* pRes = NULL;
SFirstLastRes* pRes = (SFirstLastRes*)((char*)buf + VARSTR_HEADER_SIZE);
TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
buf = taosMemoryCalloc(1, pCol->info.bytes);
TSDB_CHECK_NULL(buf, code, lino, _end, terrno);
pRes = (SFirstLastRes*)((char*)buf + VARSTR_HEADER_SIZE);
pRes->bytes = 0;
pRes->hasResult = true;
pRes->isNull = true;
varDataSetLen(buf, pCol->info.bytes - VARSTR_HEADER_SIZE);
int32_t code = colDataSetVal(pCol, row, buf, false);
taosMemoryFree(buf);
code = colDataSetVal(pCol, row, buf, false);
TSDB_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
if (buf != NULL) {
taosMemoryFreeClear(buf);
}
return code;
}
static int32_t saveOneRowForLastRaw(SLastCol* pColVal, SCacheRowsReader* pReader, const int32_t slotId,
SColumnInfoData* pColInfoData, int32_t numOfRows) {
SColVal* pVal = &pColVal->colVal;
int32_t code = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SColVal* pVal = NULL;
TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
pVal = &pColVal->colVal;
// allNullRow = false;
if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {
colDataSetNULL(pColInfoData, numOfRows);
} else {
TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);
memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);
code = colDataSetVal(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
TSDB_CHECK_CODE(code, lino, _end);
}
} else {
code = colDataSetVal(pColInfoData, numOfRows, (const char*)&pVal->value.val, !COL_VAL_IS_VALUE(pVal));
TSDB_CHECK_CODE(code, lino, _end);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
const int32_t* dstSlotIds, void** pRes, const char* idStr) {
int32_t numOfRows = pBlock->info.rows;
int32_t code = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t numOfRows = 0;
SArray* funcTypeBlockArray = NULL;
TSDB_CHECK_NULL(pBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
if (pReader->numOfCols > 0) {
TSDB_CHECK_NULL(slotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(dstSlotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(pRes, code, lino, _end, TSDB_CODE_INVALID_PARA);
}
numOfRows = pBlock->info.rows;
if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
uint64_t ts = TSKEY_MIN;
SFirstLastRes* p = NULL;
col_id_t colId = -1;
SArray* funcTypeBlockArray = taosArrayInit(pReader->numOfCols, sizeof(int32_t));
if (funcTypeBlockArray == NULL) {
return terrno;
}
funcTypeBlockArray = taosArrayInit(pReader->numOfCols, sizeof(int32_t));
TSDB_CHECK_NULL(funcTypeBlockArray, code, lino, _end, terrno);
for (int32_t i = 0; i < pReader->numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
if (pColInfoData == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
int32_t funcType = FUNCTION_TYPE_CACHE_LAST;
if (pReader->pFuncTypeList != NULL && taosArrayGetSize(pReader->pFuncTypeList) > i) {
void* pVal = taosArrayGet(pReader->pFuncTypeList, i);
if (pVal == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
funcType = *(int32_t*)pVal;
pVal = taosArrayGet(pReader->pFuncTypeList, i);
if (pVal == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
void* px = taosArrayInsert(funcTypeBlockArray, dstSlotIds[i], pVal);
if (px == NULL) {
return terrno;
}
TSDB_CHECK_NULL(px, code, lino, _end, terrno);
}
if (slotIds[i] == -1) {
@ -110,24 +137,18 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
}
code = setFirstLastResColToNull(pColInfoData, numOfRows);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
continue;
}
int32_t slotId = slotIds[i];
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
if (pColVal == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
colId = pColVal->colVal.cid;
if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
continue;
}
@ -154,22 +175,16 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
p->hasResult = true;
varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
code = colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
}
for (int32_t idx = 0; idx < taosArrayGetSize(pBlock->pDataBlock); ++idx) {
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, idx);
if (pCol == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
if (idx < funcTypeBlockArray->size) {
void* pVal = taosArrayGet(funcTypeBlockArray, idx);
if (pVal == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
int32_t funcType = *(int32_t*)pVal;
if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
@ -182,17 +197,13 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
colDataSetNULL(pCol, numOfRows);
} else {
code = colDataSetVal(pCol, numOfRows, (const char*)&ts, false);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
}
continue;
} else if (pReader->numOfCols == 1 && idx != dstSlotIds[0] && (pCol->info.colId == colId || colId == -1)) {
if (p && !p->isNull) {
code = colDataSetVal(pCol, numOfRows, p->buf, false);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
} else {
colDataSetNULL(pCol, numOfRows);
}
@ -201,13 +212,10 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
// pBlock->info.rows += allNullRow ? 0 : 1;
++pBlock->info.rows;
taosArrayDestroy(funcTypeBlockArray);
} else if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
for (int32_t i = 0; i < pReader->numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
if (pColInfoData == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
int32_t slotId = slotIds[i];
if (slotId == -1) {
@ -216,47 +224,53 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
}
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
if (pColVal == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
if (code) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
}
// pBlock->info.rows += allNullRow ? 0 : 1;
++pBlock->info.rows;
} else {
tsdbError("invalid retrieve type:%d, %s", pReader->type, idStr);
return TSDB_CODE_INVALID_PARA;
code = TSDB_CODE_INVALID_PARA;
TSDB_CHECK_CODE(code, lino, _end);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
if (funcTypeBlockArray != NULL) {
taosArrayDestroy(funcTypeBlockArray);
}
return code;
}
static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* idstr) {
int32_t numOfTables = p->numOfTables;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t numOfTables = 0;
TSDB_CHECK_NULL(p, code, lino, _end, TSDB_CODE_INVALID_PARA);
numOfTables = p->numOfTables;
if (suid != 0) {
code = metaGetTbTSchemaNotNull(p->pVnode->pMeta, suid, -1, 1, &p->pSchema);
if (TSDB_CODE_SUCCESS != code) {
tsdbWarn("stable:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", suid, idstr);
if (code == TSDB_CODE_NOT_FOUND) {
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
} else {
return code;
code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
}
TSDB_CHECK_CODE(code, lino, _end);
}
} else {
for (int32_t i = 0; i < numOfTables; ++i) {
uint64_t uid = p->pTableList[i].uid;
code = metaGetTbTSchemaMaybeNull(p->pVnode->pMeta, uid, -1, 1, &p->pSchema);
if(code != TSDB_CODE_SUCCESS) {
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
if (p->pSchema != NULL) {
break;
}
@ -267,33 +281,52 @@ static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* id
// all queried tables have been dropped already, return immediately.
if (p->pSchema == NULL) {
tsdbWarn("all queried tables has been dropped, try next group, %s", idstr);
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
TSDB_CHECK_CODE(code, lino, _end);
}
}
return TSDB_CODE_SUCCESS;
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SCacheRowsReader* pReader = (SCacheRowsReader*)reader;
TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
pReader->pTableList = pTableIdList;
pReader->numOfTables = numOfTables;
pReader->lastTs = INT64_MIN;
destroySttBlockReader(pReader->pLDataIterArray, NULL);
pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
TSDB_CHECK_NULL(pReader->pLDataIterArray, code, lino, _end, terrno);
return (pReader->pLDataIterArray != NULL) ? TSDB_CODE_SUCCESS : terrno;
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr,
SArray* pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SCacheRowsReader* p = NULL;
TSDB_CHECK_NULL(pVnode, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
*pReader = NULL;
SCacheRowsReader* p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));
if (p == NULL) {
return terrno;
}
p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));
TSDB_CHECK_NULL(p, code, lino, _end, terrno);
p->type = type;
p->pVnode = pVnode;
@ -307,12 +340,13 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
p->rowKey.numOfPKs = numOfPks;
if (numOfPks > 0) {
TSDB_CHECK_NULL(pPkCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
p->rowKey.pks[0].type = pPkCol->type;
if (IS_VAR_DATA_TYPE(pPkCol->type)) {
p->rowKey.pks[0].pData = taosMemoryCalloc(1, pPkCol->bytes);
if (p->rowKey.pks[0].pData == NULL) {
taosMemoryFree(p);
return terrno;
taosMemoryFreeClear(p);
TSDB_CHECK_NULL(p->rowKey.pks[0].pData, code, lino, _end, terrno);
}
}
@ -321,48 +355,46 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
if (numOfTables == 0) {
*pReader = p;
return TSDB_CODE_SUCCESS;
p = NULL;
goto _end;
}
p->pTableList = pTableIdList;
p->numOfTables = numOfTables;
int32_t code = setTableSchema(p, suid, idstr);
if (code != TSDB_CODE_SUCCESS) {
tsdbCacherowsReaderClose(p);
return code;
}
code = setTableSchema(p, suid, idstr);
TSDB_CHECK_CODE(code, lino, _end);
p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);
if (p->transferBuf == NULL) {
tsdbCacherowsReaderClose(p);
return terrno;
}
TSDB_CHECK_NULL(p->transferBuf, code, lino, _end, terrno);
for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
if (IS_VAR_DATA_TYPE(p->pSchema->columns[i].type)) {
p->transferBuf[i] = taosMemoryMalloc(p->pSchema->columns[i].bytes);
if (p->transferBuf[i] == NULL) {
tsdbCacherowsReaderClose(p);
return terrno;
}
TSDB_CHECK_NULL(p->transferBuf[i], code, lino, _end, terrno);
}
}
if (idstr != NULL) {
p->idstr = taosStrdup(idstr);
if (idstr != NULL && p->idstr == NULL) {
tsdbCacherowsReaderClose(p);
return terrno;
TSDB_CHECK_NULL(p->idstr, code, lino, _end, terrno);
}
code = taosThreadMutexInit(&p->readerMutex, NULL);
if (code) {
tsdbCacherowsReaderClose(p);
return code;
}
TSDB_CHECK_CODE(code, lino, _end);
p->lastTs = INT64_MIN;
*pReader = p;
p = NULL;
_end:
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
*pReader = NULL;
}
if (p != NULL) {
tsdbCacherowsReaderClose(p);
}
return code;
}
@ -393,6 +425,7 @@ void tsdbCacherowsReaderClose(void* pReader) {
if (p->pLDataIterArray) {
destroySttBlockReader(p->pLDataIterArray, NULL);
p->pLDataIterArray = NULL;
}
if (p->pFileReader) {
@ -443,39 +476,32 @@ static int32_t tsdbCacheQueryReseek(void* pQHandle) {
int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, const int32_t* dstSlotIds,
SArray* pTableUidList, bool* pGotAll) {
if (pReader == NULL || pResBlock == NULL) {
return TSDB_CODE_INVALID_PARA;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
bool hasRes = false;
SArray* pRow = NULL;
void** pRes = NULL;
SCacheRowsReader* pr = pReader;
SCacheRowsReader* pr = NULL;
int32_t pkBufLen = 0;
TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
TSDB_CHECK_NULL(pResBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
pr = pReader;
pr->pReadSnap = NULL;
pRow = taosArrayInit(TARRAY_SIZE(pr->pCidList), sizeof(SLastCol));
if (pRow == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(pRow, code, lino, _end, terrno);
pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);
if (pRes == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(pRes, code, lino, _end, terrno);
pkBufLen = (pr->rowKey.numOfPKs > 0) ? pr->pkColumn.bytes : 0;
for (int32_t j = 0; j < pr->numOfCols; ++j) {
int32_t bytes = (slotIds[j] == -1) ? 1 : pr->pSchema->columns[slotIds[j]].bytes;
pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + pkBufLen + VARSTR_HEADER_SIZE);
if (pRes[j] == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(pRes[j], code, lino, _end, terrno);
SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
p->ts = INT64_MIN;
@ -483,9 +509,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
(void)taosThreadMutexLock(&pr->readerMutex);
code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap, pr->idstr);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
TSDB_CHECK_CODE(code, lino, _end);
int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
@ -494,20 +518,14 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
// retrieve the only one last row of all tables in the uid list.
if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
SArray* pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));
if (pLastCols == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(pLastCols, code, lino, _end, terrno);
for (int32_t i = 0; i < pr->numOfCols; ++i) {
int32_t slotId = slotIds[i];
if (slotId == -1) {
SLastCol p = {.rowKey.ts = INT64_MIN, .colVal.value.type = TSDB_DATA_TYPE_BOOL, .colVal.flag = CV_FLAG_NULL};
void* px = taosArrayPush(pLastCols, &p);
if (px == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(px, code, lino, _end, terrno);
continue;
}
struct STColumn* pCol = &pr->pSchema->columns[slotId];
@ -518,29 +536,19 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
for (int32_t j = 0; j < pr->rowKey.numOfPKs; j++) {
p.rowKey.pks[j].type = pr->pkColumn.type;
if (IS_VAR_DATA_TYPE(pr->pkColumn.type)) {
p.rowKey.pks[j].pData = taosMemoryCalloc(1, pr->pkColumn.bytes);
if (p.rowKey.pks[j].pData == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(p.rowKey.pks[j].pData, code, lino, _end, terrno);
}
}
}
if (IS_VAR_DATA_TYPE(pCol->type)) {
p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));
if (p.colVal.value.pData == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(p.colVal.value.pData, code, lino, _end, terrno);
}
void* px = taosArrayPush(pLastCols, &p);
if (px == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(px, code, lino, _end, terrno);
}
int64_t st = taosGetTimestampUs();
@ -551,9 +559,8 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
if (code == -1) { // fix the invalid return code
code = 0;
} else if (code != 0) {
goto _end;
}
TSDB_CHECK_CODE(code, lino, _end);
if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
@ -600,10 +607,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
if (k == 0) {
if (TARRAY_SIZE(pTableUidList) == 0) {
void* px = taosArrayPush(pTableUidList, &uid);
if (px == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(px, code, lino, _end, terrno);
} else {
taosArraySet(pTableUidList, 0, &uid);
}
@ -654,9 +658,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
if (hasRes) {
code = saveOneRow(pLastCols, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
if (code) {
goto _end;
}
TSDB_CHECK_CODE(code, lino, _end);
}
taosArrayDestroyEx(pLastCols, tsdbCacheFreeSLastColItem);
@ -668,9 +670,8 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
if ((code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype)) != 0) {
if (code == -1) { // fix the invalid return code
code = 0;
} else if (code != 0) {
goto _end;
}
TSDB_CHECK_CODE(code, lino, _end);
}
if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
@ -679,17 +680,12 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
}
code = saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
if (code) {
goto _end;
}
TSDB_CHECK_CODE(code, lino, _end);
taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
void* px = taosArrayPush(pTableUidList, &uid);
if (px == NULL) {
code = terrno;
goto _end;
}
TSDB_CHECK_NULL(px, code, lino, _end, terrno);
++pr->tableIndex;
if (pResBlock->info.rows >= pResBlock->info.capacity) {
@ -702,6 +698,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
}
} else {
code = TSDB_CODE_INVALID_PARA;
TSDB_CHECK_CODE(code, lino, _end);
}
_end:
@ -723,5 +720,8 @@ _end:
taosMemoryFree(pRes);
taosArrayDestroy(pRow);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}

3795
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

File diff suppressed because it is too large Load Diff

1006
source/dnode/vnode/src/tsdb/tsdbReadUtil.c
View File

File diff suppressed because it is too large Load Diff

10
source/dnode/vnode/src/tsdb/tsdbReadUtil.h
View File

@ -30,13 +30,13 @@ extern "C" {
do { \
(_w)->skey = INT64_MAX; \
(_w)->ekey = INT64_MIN; \
} while (0);
} while (0)
#define INIT_KEYRANGE(_k) \
do { \
(_k)->skey.ts = INT64_MAX; \
(_k)->ekey.ts = INT64_MIN; \
} while (0);
} while (0)
#define tRowGetKeyEx(_pRow, _pKey) \
{ \
@ -72,7 +72,6 @@ typedef struct STsdbReaderInfo {
} STsdbReaderInfo;
typedef struct SBlockInfoBuf {
int32_t currentIndex;
SArray* pData;
int32_t numPerBucket;
int32_t numOfTables;
@ -241,7 +240,6 @@ typedef struct SDataBlockIter {
int32_t index;
SArray* blockList; // SArray<SFileDataBlockInfo>
int32_t order;
SDataBlk block; // current SDataBlk data
} SDataBlockIter;
typedef struct SFileBlockDumpInfo {
@ -321,7 +319,7 @@ int32_t createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf, c
int32_t initTableBlockScanInfo(STableBlockScanInfo* pScanInfo, uint64_t uid, SSHashObj* pTableMap,
STsdbReader* pReader);
void clearBlockScanInfo(STableBlockScanInfo* p);
void destroyAllBlockScanInfo(SSHashObj* pTableMap);
void destroyAllBlockScanInfo(SSHashObj** pTableMap);
void resetAllDataBlockScanInfo(SSHashObj* pTableMap, int64_t ts, int32_t step);
void cleanupInfoForNextFileset(SSHashObj* pTableMap);
int32_t ensureBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables);
@ -335,7 +333,7 @@ void clearBrinBlockIter(SBrinRecordIter* pIter);
// initialize block iterator API
int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIter, int32_t numOfBlocks, SArray* pTableList);
bool blockIteratorNext(SDataBlockIter* pBlockIter, const char* idStr);
bool blockIteratorNext(SDataBlockIter* pBlockIter);
// load tomb data API (stt/mem only for one table each, tomb data from data files are load for all tables at one time)
int32_t loadMemTombData(SArray** ppMemDelData, STbData* pMemTbData, STbData* piMemTbData, int64_t ver);

14
source/libs/executor/src/sysscanoperator.c
View File

@ -685,11 +685,11 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
pAPI->metaFn.pauseTableMetaCursor(pInfo->pCur);
break;
}
} else {
}
// if pInfo->pRes->info.rows == 0, also need to add the meta to pDataBlock
code = sysTableUserColsFillOneTableCols(pInfo, dbname, &numOfRows, pDataBlock, tableName, schemaRow, typeName);
QUERY_CHECK_CODE(code, lino, _end);
}
}
if (numOfRows > 0) {
pAPI->metaFn.pauseTableMetaCursor(pInfo->pCur);
@ -761,7 +761,7 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
SMetaReader smrChildTable = {0};
pAPI->metaReaderFn.initReader(&smrChildTable, pInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);
int32_t code = pAPI->metaReaderFn.getTableEntryByName(&smrChildTable, condTableName);
code = pAPI->metaReaderFn.getTableEntryByName(&smrChildTable, condTableName);
if (code != TSDB_CODE_SUCCESS) {
// terrno has been set by pAPI->metaReaderFn.getTableEntryByName, therefore, return directly
pAPI->metaReaderFn.clearReader(&smrChildTable);
@ -847,7 +847,8 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
pAPI->metaReaderFn.clearReader(&smrSuperTable);
break;
}
} else {
}
// if pInfo->pRes->info.rows == 0, also need to add this meta into datablock.
code = sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows,
dataBlock);
if (code != TSDB_CODE_SUCCESS) {
@ -859,7 +860,6 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
dataBlock = NULL;
T_LONG_JMP(pTaskInfo->env, terrno);
}
}
pAPI->metaReaderFn.clearReader(&smrSuperTable);
}
@ -2792,7 +2792,9 @@ static int32_t doBlockInfoScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes
code = pAPI->tsdReader.tsdReaderGetDataBlockDistInfo(pBlockScanInfo->pHandle, &blockDistInfo);
QUERY_CHECK_CODE(code, lino, _end);
blockDistInfo.numOfInmemRows = (int32_t)pAPI->tsdReader.tsdReaderGetNumOfInMemRows(pBlockScanInfo->pHandle);
blockDistInfo.numOfInmemRows = 0;
code = pAPI->tsdReader.tsdReaderGetNumOfInMemRows(pBlockScanInfo->pHandle, &blockDistInfo.numOfInmemRows);
QUERY_CHECK_CODE(code, lino, _end);
SSDataBlock* pBlock = pBlockScanInfo->pResBlock;

2
source/libs/function/inc/tpercentile.h
View File

@ -26,7 +26,7 @@ extern "C" {
struct tMemBucket;
int32_t tMemBucketCreate(int32_t nElemSize, int16_t dataType, double minval, double maxval, bool hasWindowOrGroup,
struct tMemBucket **pBucket);
struct tMemBucket **pBucket, int32_t numOfElements);
void tMemBucketDestroy(struct tMemBucket **pBucket);

2
source/libs/function/src/builtinsimpl.c
View File

@ -1805,7 +1805,7 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
pResInfo->complete = true;
return TSDB_CODE_SUCCESS;
} else {
code = tMemBucketCreate(pCol->info.bytes, type, pInfo->minval, pInfo->maxval, pCtx->hasWindowOrGroup, &pInfo->pMemBucket);
code = tMemBucketCreate(pCol->info.bytes, type, pInfo->minval, pInfo->maxval, pCtx->hasWindowOrGroup, &pInfo->pMemBucket, pInfo->numOfElems);
if (TSDB_CODE_SUCCESS != code) {
return code;
}

10
source/libs/function/src/tpercentile.c
View File

@ -269,18 +269,16 @@ static void resetSlotInfo(tMemBucket *pBucket) {
}
int32_t tMemBucketCreate(int32_t nElemSize, int16_t dataType, double minval, double maxval, bool hasWindowOrGroup,
tMemBucket **pBucket) {
tMemBucket **pBucket, int32_t numOfElements) {
*pBucket = (tMemBucket *)taosMemoryCalloc(1, sizeof(tMemBucket));
if (*pBucket == NULL) {
return terrno;
}
if (hasWindowOrGroup) {
// With window or group by, we need to shrink page size and reduce page num to save memory.
(*pBucket)->numOfSlots = DEFAULT_NUM_OF_SLOT / 8 ; // 128 bucket
// With window or group by, we need to shrink page size to save memory.
(*pBucket)->bufPageSize = 4096; // 4k per page
} else {
(*pBucket)->numOfSlots = DEFAULT_NUM_OF_SLOT;
(*pBucket)->bufPageSize = 16384 * 4; // 16k per page
}
@ -302,6 +300,8 @@ int32_t tMemBucketCreate(int32_t nElemSize, int16_t dataType, double minval, dou
}
(*pBucket)->elemPerPage = ((*pBucket)->bufPageSize - sizeof(SFilePage)) / (*pBucket)->bytes;
(*pBucket)->numOfSlots = TMIN((int16_t)(numOfElements / ((*pBucket)->elemPerPage * 6)) + 1, DEFAULT_NUM_OF_SLOT);
(*pBucket)->comparFn = getKeyComparFunc((*pBucket)->type, TSDB_ORDER_ASC);
(*pBucket)->hashFunc = getHashFunc((*pBucket)->type);
@ -587,7 +587,7 @@ int32_t getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction
// try next round
tMemBucket *tmpBucket = NULL;
int32_t code = tMemBucketCreate(pMemBucket->bytes, pMemBucket->type, pSlot->range.dMinVal, pSlot->range.dMaxVal,
false, &tmpBucket);
false, &tmpBucket, pSlot->info.size);
if (TSDB_CODE_SUCCESS != code) {
tMemBucketDestroy(&tmpBucket);
return code;

1
source/libs/sync/inc/syncSnapshot.h
View File

@ -91,6 +91,7 @@ typedef struct SSyncSnapshotReceiver {
// update when begin
void *pWriter;
TdThreadMutex writerMutex;
SSnapshotParam snapshotParam;
SSnapshot snapshot;

25
source/libs/sync/src/syncSnapshot.c
View File

@ -429,6 +429,12 @@ int32_t snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId, SSyncSnapsh
pReceiver->startTime = 0;
pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN;
pReceiver->pWriter = NULL;
code = taosThreadMutexInit(&pReceiver->writerMutex, NULL);
if (code != 0) {
taosMemoryFree(pReceiver);
pReceiver = NULL;
TAOS_RETURN(code);
}
pReceiver->pSyncNode = pSyncNode;
pReceiver->fromId = fromId;
pReceiver->term = raftStoreGetTerm(pSyncNode);
@ -440,6 +446,10 @@ int32_t snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId, SSyncSnapsh
SSyncSnapBuffer *pRcvBuf = NULL;
code = syncSnapBufferCreate(&pRcvBuf);
if (pRcvBuf == NULL) {
int32_t ret = taosThreadMutexDestroy(&pReceiver->writerMutex);
if (ret != 0) {
sError("failed to destroy mutex since %s", tstrerror(ret));
}
taosMemoryFree(pReceiver);
pReceiver = NULL;
TAOS_RETURN(code);
@ -468,6 +478,7 @@ static int32_t snapshotReceiverClearInfoData(SSyncSnapshotReceiver *pReceiver) {
void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
if (pReceiver == NULL) return;
(void)taosThreadMutexLock(&pReceiver->writerMutex);
// close writer
if (pReceiver->pWriter != NULL) {
int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
@ -478,6 +489,9 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
}
pReceiver->pWriter = NULL;
}
(void)taosThreadMutexUnlock(&pReceiver->writerMutex);
(void)taosThreadMutexDestroy(&pReceiver->writerMutex);
// free snap buf
if (pReceiver->pRcvBuf) {
@ -556,7 +570,8 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
int8_t stopped = !atomic_val_compare_exchange_8(&pReceiver->start, true, false);
if (stopped) return;
(void)taosThreadMutexLock(&pReceiver->pRcvBuf->mutex);
(void)taosThreadMutexLock(&pReceiver->writerMutex);
{
if (pReceiver->pWriter != NULL) {
int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
@ -568,7 +583,11 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
} else {
sRInfo(pReceiver, "snapshot receiver stop, writer is null");
}
}
(void)taosThreadMutexUnlock(&pReceiver->writerMutex);
(void)taosThreadMutexLock(&pReceiver->pRcvBuf->mutex);
{
syncSnapBufferReset(pReceiver->pRcvBuf);
(void)snapshotReceiverClearInfoData(pReceiver);
@ -600,6 +619,8 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
raftStoreSetTerm(pReceiver->pSyncNode, pReceiver->snapshot.lastApplyTerm);
}
(void)taosThreadMutexLock(&pReceiver->writerMutex);
if (pReceiver->pWriter != NULL) {
// stop writer, apply data
code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true,
&pReceiver->snapshot);
@ -609,6 +630,8 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
}
pReceiver->pWriter = NULL;
sRInfo(pReceiver, "snapshot receiver write stopped");
}
(void)taosThreadMutexUnlock(&pReceiver->writerMutex);
// update progress
pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;

1
tests/parallel_test/cases.task
View File

@ -327,6 +327,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TS-3311.py
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TS-3821.py
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TS-5130.py
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TS-5580.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/balance_vgroups_r1.py -N 6
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/taosShell.py

118
tests/system-test/99-TDcase/TS-5580.py Normal file
View File

@ -0,0 +1,118 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import time
from util.log import *
from util.cases import *
from util.sql import *
from util.common import *
from util.sqlset import *
class TDTestCase:
updatecfgDict = {'qDebugFlag':135 , 'mDebugFlag':135}
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), True)
self.setsql = TDSetSql()
self.dbname = 'db'
self.stbname = 'stb'
self.binary_length = 20 # the length of binary for column_dict
self.nchar_length = 20 # the length of nchar for column_dict
self.ts = 1537146000000
self.column_dict = {
'ts' : 'timestamp',
'col1': 'tinyint',
'col2': 'smallint',
'col3': 'int',
'col4': 'bigint',
'col5': 'float',
'col6': 'double',
'col7': 'double',
'col8': 'double',
'col9': 'double',
'col10': 'double',
'col11': 'double',
'col12': 'double',
'col13': 'double',
'col14': 'double',
'col15': 'double',
'col16': 'double',
'col17': 'double',
'col18': 'double',
'col19': 'double'
}
self.tbnum = 500
self.rowNum = 10
self.tag_dict = {
't0':'int',
't1':'bigint',
't2':'float',
't3':'double',
't4':'bool',
't5':'bool',
't6':'bool',
't7':'bool',
't8':'bool',
't9':'bool',
't10':'bool',
't11':'bool',
't12':'bool',
't13':'bool',
't14':'bool',
't15':'bool',
't16':'bool',
't17':'bool',
't18':'bool',
't19':'bool',
}
self.tag_values = [
f'1','1','1','1','true','true','true','true','true','true','true','true','true','true','true','true','true',
'true','true','true'
]
def prepare_data(self):
tdSql.execute(f"create database if not exists {self.dbname} vgroups 2")
tdSql.execute(f'use {self.dbname}')
tdSql.execute(self.setsql.set_create_stable_sql(self.stbname,self.column_dict,self.tag_dict))
for i in range(self.tbnum):
tdSql.execute(f"create table {self.stbname}_{i} using {self.stbname} tags({self.tag_values[0]}, {self.tag_values[1]}, "
f"{self.tag_values[2]}, {self.tag_values[3]}, {self.tag_values[4]}, {self.tag_values[5]}, "
f"{self.tag_values[6]}, {self.tag_values[7]}, {self.tag_values[8]}, {self.tag_values[9]}, "
f"{self.tag_values[10]}, {self.tag_values[11]}, {self.tag_values[12]}, {self.tag_values[13]}, "
f"{self.tag_values[14]}, {self.tag_values[15]}, {self.tag_values[16]}, {self.tag_values[17]}, "
f"{self.tag_values[18]}, {self.tag_values[19]})")
def test_query_ins_tags(self):
for i in range(self.tbnum):
sql = f'select tag_name, tag_value from information_schema.ins_tags where table_name = "{self.stbname}_{i}"'
tdSql.query(sql)
tdSql.checkRows(20)
def test_query_ins_columns(self):
for i in range(self.tbnum):
sql = f'select col_name from information_schema.ins_columns where table_name = "{self.stbname}_{i}"'
tdSql.query(sql)
tdSql.checkRows(20)
def run(self):
self.prepare_data()
self.test_query_ins_tags()
self.test_query_ins_columns()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())