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

Merge branch 'main' into fix/TD-24910-M

This commit is contained in:
kailixu 2023-06-26 14:32:10 +08:00
parent 7d2a2fbebc 701e09f4ea
commit a9f49ff985
19 changed files with 376 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
cmake/cmake.options
View File

@ -189,3 +189,9 @@ option(
"If build release version" "If build release version"
OFF OFF
) )
option(
BUILD_CONTRIB
"If build thirdpart from source"
OFF
)

41
cmake/rocksdb_CMakeLists.txt.in
View File

@ -1,16 +1,29 @@
# rocksdb # rocksdb
IF (NOT ${TD_LINUX}) if (${BUILD_CONTRIB})
ExternalProject_Add(rocksdb
ExternalProject_Add(rocksdb URL https://github.com/facebook/rocksdb/archive/refs/tags/v8.1.1.tar.gz
URL https://github.com/facebook/rocksdb/archive/refs/tags/v8.1.1.tar.gz URL_HASH MD5=3b4c97ee45df9c8a5517308d31ab008b
URL_HASH MD5=3b4c97ee45df9c8a5517308d31ab008b DOWNLOAD_NO_PROGRESS 1
DOWNLOAD_NO_PROGRESS 1 DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download" SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb"
SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb" CONFIGURE_COMMAND ""
CONFIGURE_COMMAND "" BUILD_COMMAND ""
BUILD_COMMAND "" INSTALL_COMMAND ""
INSTALL_COMMAND "" TEST_COMMAND ""
TEST_COMMAND "" )
) else()
ENDIF(NOT ${TD_LINUX}) if (NOT ${TD_LINUX})
ExternalProject_Add(rocksdb
URL https://github.com/facebook/rocksdb/archive/refs/tags/v8.1.1.tar.gz
URL_HASH MD5=3b4c97ee45df9c8a5517308d31ab008b
DOWNLOAD_NO_PROGRESS 1
DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
endif()
endif()

2
cmake/taosadapter_CMakeLists.txt.in
View File

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

169
contrib/CMakeLists.txt
View File

@ -77,19 +77,25 @@ if(${BUILD_WITH_LEVELDB})
cat("${TD_SUPPORT_DIR}/leveldb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) cat("${TD_SUPPORT_DIR}/leveldb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
endif(${BUILD_WITH_LEVELDB}) endif(${BUILD_WITH_LEVELDB})
# rocksdb
IF (NOT ${TD_LINUX})
if(${BUILD_WITH_ROCKSDB})
cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
add_definitions(-DUSE_ROCKSDB)
endif(${BUILD_WITH_ROCKSDB})
ELSE()
if(${BUILD_WITH_ROCKSDB})
#cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
add_definitions(-DUSE_ROCKSDB)
endif(${BUILD_WITH_ROCKSDB})
ENDIF(NOT ${TD_LINUX}) if (${BUILD_CONTRIB})
if(${BUILD_WITH_ROCKSDB})
cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
add_definitions(-DUSE_ROCKSDB)
endif()
else()
if (NOT ${TD_LINUX})
if(${BUILD_WITH_ROCKSDB})
cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
add_definitions(-DUSE_ROCKSDB)
endif(${BUILD_WITH_ROCKSDB})
else()
if(${BUILD_WITH_ROCKSDB})
add_definitions(-DUSE_ROCKSDB)
endif(${BUILD_WITH_ROCKSDB})
endif()
endif()
# canonical-raft # canonical-raft
if(${BUILD_WITH_CRAFT}) if(${BUILD_WITH_CRAFT})
@ -245,68 +251,103 @@ if (${BUILD_WITH_UV})
endif(${TD_LINUX}) endif(${TD_LINUX})
endif (${BUILD_WITH_UV}) endif (${BUILD_WITH_UV})
IF (NOT ${TD_LINUX}) if (${BUILD_WITH_ROCKSDB})
if(${BUILD_WITH_ROCKSDB}) if (${BUILD_CONTRIB})
if(${TD_LINUX}) if(${TD_LINUX})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL} -Wno-error=maybe-uninitialized -Wno-error=unused-but-set-variable -Wno-error=unused-variable -Wno-error=unused-function -Wno-errno=unused-private-field -Wno-error=unused-result") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL} -Wno-error=maybe-uninitialized -Wno-error=unused-but-set-variable -Wno-error=unused-variable -Wno-error=unused-function -Wno-errno=unused-private-field -Wno-error=unused-result")
IF ("${CMAKE_BUILD_TYPE}" STREQUAL "") if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
SET(CMAKE_BUILD_TYPE Release) SET(CMAKE_BUILD_TYPE Release)
endif() endif()
endif(${TD_LINUX}) endif(${TD_LINUX})
MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS}) MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
if(${TD_DARWIN}) if(${TD_DARWIN})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
endif(${TD_DARWIN}) endif(${TD_DARWIN})
if (${TD_DARWIN_ARM64}) if (${TD_DARWIN_ARM64})
set(HAS_ARMV8_CRC true) set(HAS_ARMV8_CRC true)
endif(${TD_DARWIN_ARM64}) endif(${TD_DARWIN_ARM64})
if (${TD_WINDOWS}) if (${TD_WINDOWS})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
endif(${TD_WINDOWS}) option(WITH_JNI "" OFF)
option(WITH_MD_LIBRARY "build with MD" OFF)
set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
endif(${TD_WINDOWS})
if(${TD_DARWIN}) if(${TD_DARWIN})
option(HAVE_THREAD_LOCAL "" OFF) option(HAVE_THREAD_LOCAL "" OFF)
option(WITH_IOSTATS_CONTEXT "" OFF) option(WITH_IOSTATS_CONTEXT "" OFF)
option(WITH_PERF_CONTEXT "" OFF) option(WITH_PERF_CONTEXT "" OFF)
endif(${TD_DARWIN}) endif(${TD_DARWIN})
if(${TD_WINDOWS}) option(WITH_FALLOCATE "" OFF)
option(WITH_JNI "" OFF) option(WITH_JEMALLOC "" OFF)
endif(${TD_WINDOWS}) option(WITH_GFLAGS "" OFF)
option(PORTABLE "" ON)
option(WITH_LIBURING "" OFF)
option(FAIL_ON_WARNINGS OFF)
if(${TD_WINDOWS}) option(WITH_TESTS "" OFF)
option(WITH_MD_LIBRARY "build with MD" OFF) option(WITH_BENCHMARK_TOOLS "" OFF)
set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib) option(WITH_TOOLS "" OFF)
endif(${TD_WINDOWS}) option(WITH_LIBURING "" OFF)
option(ROCKSDB_BUILD_SHARED "Build shared versions of the RocksDB libraries" OFF)
add_subdirectory(rocksdb EXCLUDE_FROM_ALL)
target_include_directories(
rocksdb
PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/rocksdb/include>
)
else()
if (NOT ${TD_LINUX})
MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
if(${TD_DARWIN})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
endif(${TD_DARWIN})
if (${TD_DARWIN_ARM64})
set(HAS_ARMV8_CRC true)
endif(${TD_DARWIN_ARM64})
if (${TD_WINDOWS})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
option(WITH_JNI "" OFF)
option(WITH_MD_LIBRARY "build with MD" OFF)
set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
endif(${TD_WINDOWS})
option(WITH_FALLOCATE "" OFF) if(${TD_DARWIN})
option(WITH_JEMALLOC "" OFF) option(HAVE_THREAD_LOCAL "" OFF)
option(WITH_GFLAGS "" OFF) option(WITH_IOSTATS_CONTEXT "" OFF)
option(PORTABLE "" ON) option(WITH_PERF_CONTEXT "" OFF)
option(WITH_LIBURING "" OFF) endif(${TD_DARWIN})
option(FAIL_ON_WARNINGS OFF)
option(WITH_TESTS "" OFF) option(WITH_FALLOCATE "" OFF)
option(WITH_BENCHMARK_TOOLS "" OFF) option(WITH_JEMALLOC "" OFF)
option(WITH_TOOLS "" OFF) option(WITH_GFLAGS "" OFF)
option(WITH_LIBURING "" OFF) option(PORTABLE "" ON)
IF (TD_LINUX) option(WITH_LIBURING "" OFF)
option(ROCKSDB_BUILD_SHARED "Build shared versions of the RocksDB libraries" OFF) option(FAIL_ON_WARNINGS OFF)
ELSE()
option(ROCKSDB_BUILD_SHARED "Build shared versions of the RocksDB libraries" OFF) option(WITH_TESTS "" OFF)
ENDIF() option(WITH_BENCHMARK_TOOLS "" OFF)
add_subdirectory(rocksdb EXCLUDE_FROM_ALL) option(WITH_TOOLS "" OFF)
target_include_directories( option(WITH_LIBURING "" OFF)
rocksdb
PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/rocksdb/include> option(ROCKSDB_BUILD_SHARED "Build shared versions of the RocksDB libraries" OFF)
) add_subdirectory(rocksdb EXCLUDE_FROM_ALL)
endif(${BUILD_WITH_ROCKSDB}) target_include_directories(
ENDIF (NOT ${TD_LINUX}) rocksdb
PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/rocksdb/include>
)
endif()
endif()
endif()
# lucene # lucene
# To support build on ubuntu: sudo apt-get install libboost-all-dev # To support build on ubuntu: sudo apt-get install libboost-all-dev

2
include/util/tlosertree.h
View File

@ -43,7 +43,7 @@ typedef struct SMultiwayMergeTreeInfo {
int32_t tMergeTreeCreate(SMultiwayMergeTreeInfo **pTree, uint32_t numOfEntries, void *param, int32_t tMergeTreeCreate(SMultiwayMergeTreeInfo **pTree, uint32_t numOfEntries, void *param,
__merge_compare_fn_t compareFn); __merge_compare_fn_t compareFn);
void tMergeTreeDestroy(SMultiwayMergeTreeInfo *pTree); void tMergeTreeDestroy(SMultiwayMergeTreeInfo **pTree);
void tMergeTreeAdjust(SMultiwayMergeTreeInfo *pTree, int32_t idx); void tMergeTreeAdjust(SMultiwayMergeTreeInfo *pTree, int32_t idx);

12
source/client/src/clientImpl.c
View File

@ -1699,6 +1699,7 @@ static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, i
len += lenTmp; len += lenTmp;
pStart += lenTmp; pStart += lenTmp;
int32_t estimateColLen = 0;
for (int32_t j = 0; j < numOfRows; ++j) { for (int32_t j = 0; j < numOfRows; ++j) {
if (offset[j] == -1) { if (offset[j] == -1) {
continue; continue;
@ -1708,20 +1709,21 @@ static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, i
int32_t jsonInnerType = *data; int32_t jsonInnerType = *data;
char* jsonInnerData = data + CHAR_BYTES; char* jsonInnerData = data + CHAR_BYTES;
if (jsonInnerType == TSDB_DATA_TYPE_NULL) { if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
len += (VARSTR_HEADER_SIZE + strlen(TSDB_DATA_NULL_STR_L)); estimateColLen += (VARSTR_HEADER_SIZE + strlen(TSDB_DATA_NULL_STR_L));
} else if (tTagIsJson(data)) { } else if (tTagIsJson(data)) {
len += (VARSTR_HEADER_SIZE + ((const STag*)(data))->len); estimateColLen += (VARSTR_HEADER_SIZE + ((const STag*)(data))->len);
} else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) { // value -> "value" } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) { // value -> "value"
len += varDataTLen(jsonInnerData) + CHAR_BYTES * 2; estimateColLen += varDataTLen(jsonInnerData) + CHAR_BYTES * 2;
} else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) { } else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
len += (VARSTR_HEADER_SIZE + 32); estimateColLen += (VARSTR_HEADER_SIZE + 32);
} else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) { } else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) {
len += (VARSTR_HEADER_SIZE + 5); estimateColLen += (VARSTR_HEADER_SIZE + 5);
} else { } else {
tscError("estimateJsonLen error: invalid type:%d", jsonInnerType); tscError("estimateJsonLen error: invalid type:%d", jsonInnerType);
return -1; return -1;
} }
} }
len += TMAX(colLen, estimateColLen);
} else if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) { } else if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
int32_t lenTmp = numOfRows * sizeof(int32_t); int32_t lenTmp = numOfRows * sizeof(int32_t);
len += (lenTmp + colLen); len += (lenTmp + colLen);

5
source/common/src/tdatablock.c
View File

@ -1713,7 +1713,8 @@ static int32_t colDataMoveVarData(SColumnInfoData* pColInfoData, size_t start, s
static void colDataTrimFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) { static void colDataTrimFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) {
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, n, total); // pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, n, total);
memmove(pColInfoData->varmeta.offset, &pColInfoData->varmeta.offset[n], (total - n) * sizeof(int32_t));
// clear the offset value of the unused entries. // clear the offset value of the unused entries.
memset(&pColInfoData->varmeta.offset[total - n], 0, n); memset(&pColInfoData->varmeta.offset[total - n], 0, n);
@ -1745,7 +1746,7 @@ int32_t blockDataTrimFirstRows(SSDataBlock* pBlock, size_t n) {
static void colDataKeepFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) { static void colDataKeepFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) {
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, 0, n); // pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, 0, n);
memset(&pColInfoData->varmeta.offset[n], 0, total - n); memset(&pColInfoData->varmeta.offset[n], 0, total - n);
} else { // reset the bitmap value } else { // reset the bitmap value
/*int32_t stopIndex = BitmapLen(n) * 8; /*int32_t stopIndex = BitmapLen(n) * 8;

86
source/dnode/vnode/CMakeLists.txt
View File

@ -80,34 +80,50 @@ IF (TD_VNODE_PLUGINS)
) )
ENDIF () ENDIF ()
IF (NOT ${TD_LINUX}) # IF (NOT ${TD_LINUX})
target_include_directories( # target_include_directories(
vnode # vnode
PUBLIC "inc" # PUBLIC "inc"
PUBLIC "src/inc" # PUBLIC "src/inc"
PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar" # PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include" # PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include"
) # )
ELSE() # ELSE()
target_include_directories( # target_include_directories(
vnode # vnode
PUBLIC "inc" # PUBLIC "inc"
PUBLIC "src/inc" # PUBLIC "src/inc"
PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar" # PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
) # )
#ENDIF(NOT ${TD_LINUX})
ENDIF(NOT ${TD_LINUX}) if (${BUILD_CONTRIB})
IF (TD_LINUX)
target_include_directories( target_include_directories(
vnode vnode
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static" PUBLIC "inc"
PUBLIC "src/inc"
PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include"
) )
else()
target_include_directories(
vnode
PUBLIC "inc"
PUBLIC "src/inc"
PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
)
if (${TD_LINUX})
target_include_directories(
vnode
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static"
)
target_link_directories(
vnode
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static"
)
endif()
endif()
target_link_directories(
vnode
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static"
)
target_link_libraries( target_link_libraries(
vnode vnode
PUBLIC os PUBLIC os
@ -128,28 +144,6 @@ target_link_libraries(
PUBLIC stream PUBLIC stream
PUBLIC index PUBLIC index
) )
ELSE()
target_link_libraries(
vnode
PUBLIC os
PUBLIC util
PUBLIC common
PUBLIC tfs
PUBLIC wal
PUBLIC qworker
PUBLIC sync
PUBLIC executor
PUBLIC scheduler
PUBLIC tdb
# PUBLIC bdb
# PUBLIC scalar
PUBLIC rocksdb
PUBLIC transport
PUBLIC stream
PUBLIC index
)
ENDIF()
IF (TD_GRANT) IF (TD_GRANT)
TARGET_LINK_LIBRARIES(vnode PUBLIC grant) TARGET_LINK_LIBRARIES(vnode PUBLIC grant)

4
source/libs/executor/inc/tsort.h
View File

@ -170,6 +170,10 @@ SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle);
*/ */
int32_t getProperSortPageSize(size_t rowSize, uint32_t numOfCols); int32_t getProperSortPageSize(size_t rowSize, uint32_t numOfCols);
bool tsortIsClosed(SSortHandle* pHandle);
void tsortSetClosed(SSortHandle* pHandle);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

3
source/libs/executor/src/executil.c
View File

@ -1133,8 +1133,7 @@ int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, S
memcpy(pPayload + sizeof(int32_t), taosArrayGet(pUidList, 0), numOfTables * sizeof(uint64_t)); memcpy(pPayload + sizeof(int32_t), taosArrayGet(pUidList, 0), numOfTables * sizeof(uint64_t));
} }
// metaUidFilterCachePut(metaHandle, pScanNode->suid, context.digest, tListLen(context.digest), pPayload, pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest), pPayload, size, 1);
// size, 1);
digest[0] = 1; digest[0] = 1;
memcpy(digest + 1, context.digest, tListLen(context.digest)); memcpy(digest + 1, context.digest, tListLen(context.digest));
} }

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

@ -2895,6 +2895,11 @@ SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock*
} }
} }
if (tsortIsClosed(pHandle)) {
terrno = TSDB_CODE_TSC_QUERY_CANCELLED;
T_LONG_JMP(pOperator->pTaskInfo->env, terrno);
}
bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo); bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);
qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows, qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,
pInfo->limitInfo.numOfOutputRows); pInfo->limitInfo.numOfOutputRows);

10
source/libs/executor/src/sortoperator.c
View File

@ -228,6 +228,11 @@ SSDataBlock* doSort(SOperatorInfo* pOperator) {
// multi-group case not handle here // multi-group case not handle here
SSDataBlock* pBlock = NULL; SSDataBlock* pBlock = NULL;
while (1) { while (1) {
if (tsortIsClosed(pInfo->pSortHandle)) {
terrno = TSDB_CODE_TSC_QUERY_CANCELLED;
T_LONG_JMP(pOperator->pTaskInfo->env, terrno);
}
pBlock = getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pBlock = getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity,
pInfo->matchInfo.pList, pInfo); pInfo->matchInfo.pList, pInfo);
if (pBlock == NULL) { if (pBlock == NULL) {
@ -439,6 +444,11 @@ SSDataBlock* doGroupSort(SOperatorInfo* pOperator) {
SSDataBlock* pBlock = NULL; SSDataBlock* pBlock = NULL;
while (pInfo->pCurrSortHandle != NULL) { while (pInfo->pCurrSortHandle != NULL) {
if (tsortIsClosed(pInfo->pCurrSortHandle)) {
terrno = TSDB_CODE_TSC_QUERY_CANCELLED;
T_LONG_JMP(pOperator->pTaskInfo->env, terrno);
}
// beginSortGroup would fetch all child blocks of pInfo->currGroupId; // beginSortGroup would fetch all child blocks of pInfo->currGroupId;
ASSERT(pInfo->childOpStatus != CHILD_OP_SAME_GROUP); ASSERT(pInfo->childOpStatus != CHILD_OP_SAME_GROUP);
pBlock = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pBlock = getGroupSortedBlockData(pInfo->pCurrSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity,

24
source/libs/executor/src/tsort.c
View File

@ -46,6 +46,7 @@ struct SSortHandle {
SMsortComparParam cmpParam; SMsortComparParam cmpParam;
int32_t numOfCompletedSources; int32_t numOfCompletedSources;
bool opened; bool opened;
int8_t closed;
const char* idStr; const char* idStr;
bool inMemSort; bool inMemSort;
bool needAdjust; bool needAdjust;
@ -152,7 +153,7 @@ void tsortDestroySortHandle(SSortHandle* pSortHandle) {
tsortClose(pSortHandle); tsortClose(pSortHandle);
if (pSortHandle->pMergeTree != NULL) { if (pSortHandle->pMergeTree != NULL) {
tMergeTreeDestroy(pSortHandle->pMergeTree); tMergeTreeDestroy(&pSortHandle->pMergeTree);
} }
destroyDiskbasedBuf(pSortHandle->pBuf); destroyDiskbasedBuf(pSortHandle->pBuf);
@ -581,6 +582,11 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
SArray* pPageIdList = taosArrayInit(4, sizeof(int32_t)); SArray* pPageIdList = taosArrayInit(4, sizeof(int32_t));
while (1) { while (1) {
if (tsortIsClosed(pHandle)) {
code = terrno = TSDB_CODE_TSC_QUERY_CANCELLED;
return code;
}
SSDataBlock* pDataBlock = getSortedBlockDataInner(pHandle, &pHandle->cmpParam, numOfRows); SSDataBlock* pDataBlock = getSortedBlockDataInner(pHandle, &pHandle->cmpParam, numOfRows);
if (pDataBlock == NULL) { if (pDataBlock == NULL) {
break; break;
@ -609,7 +615,7 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
} }
sortComparCleanup(&pHandle->cmpParam); sortComparCleanup(&pHandle->cmpParam);
tMergeTreeDestroy(pHandle->pMergeTree); tMergeTreeDestroy(&pHandle->pMergeTree);
pHandle->numOfCompletedSources = 0; pHandle->numOfCompletedSources = 0;
SSDataBlock* pBlock = createOneDataBlock(pHandle->pDataBlock, false); SSDataBlock* pBlock = createOneDataBlock(pHandle->pDataBlock, false);
@ -803,10 +809,19 @@ int32_t tsortOpen(SSortHandle* pHandle) {
} }
int32_t tsortClose(SSortHandle* pHandle) { int32_t tsortClose(SSortHandle* pHandle) {
// do nothing atomic_val_compare_exchange_8(&pHandle->closed, 0, 1);
taosMsleep(10);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
bool tsortIsClosed(SSortHandle* pHandle) {
return atomic_val_compare_exchange_8(&pHandle->closed, 1, 2);
}
void tsortSetClosed(SSortHandle* pHandle) {
atomic_store_8(&pHandle->closed, 2);
}
int32_t tsortSetFetchRawDataFp(SSortHandle* pHandle, _sort_fetch_block_fn_t fetchFp, void (*fp)(SSDataBlock*, void*), int32_t tsortSetFetchRawDataFp(SSortHandle* pHandle, _sort_fetch_block_fn_t fetchFp, void (*fp)(SSDataBlock*, void*),
void* param) { void* param) {
pHandle->fetchfp = fetchFp; pHandle->fetchfp = fetchFp;
@ -826,6 +841,9 @@ int32_t tsortSetCompareGroupId(SSortHandle* pHandle, bool compareGroupId) {
} }
STupleHandle* tsortNextTuple(SSortHandle* pHandle) { STupleHandle* tsortNextTuple(SSortHandle* pHandle) {
if (tsortIsClosed(pHandle)) {
return NULL;
}
if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) { if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) {
return NULL; return NULL;
} }

24
source/libs/parser/src/parAuthenticator.c
View File

@ -28,6 +28,10 @@ typedef struct SSelectAuthCxt {
SSelectStmt* pSelect; SSelectStmt* pSelect;
} SSelectAuthCxt; } SSelectAuthCxt;
typedef struct SAuthRewriteCxt {
STableNode* pTarget;
} SAuthRewriteCxt;
static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt); static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt);
static void setUserAuthInfo(SParseContext* pCxt, const char* pDbName, const char* pTabName, AUTH_TYPE type, static void setUserAuthInfo(SParseContext* pCxt, const char* pDbName, const char* pTabName, AUTH_TYPE type,
@ -90,12 +94,26 @@ static int32_t mergeStableTagCond(SNode** pWhere, SNode* pTagCond) {
return code; return code;
} }
static int32_t appendStableTagCond(SNode** pWhere, SNode* pTagCond) { EDealRes rewriteAuthTable(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
SAuthRewriteCxt* pCxt = (SAuthRewriteCxt*)pContext;
strcpy(pCol->tableName, pCxt->pTarget->tableName);
strcpy(pCol->tableAlias, pCxt->pTarget->tableAlias);
}
return DEAL_RES_CONTINUE;
}
static int32_t rewriteAppendStableTagCond(SNode** pWhere, SNode* pTagCond, STableNode* pTable) {
SNode* pTagCondCopy = nodesCloneNode(pTagCond); SNode* pTagCondCopy = nodesCloneNode(pTagCond);
if (NULL == pTagCondCopy) { if (NULL == pTagCondCopy) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
SAuthRewriteCxt cxt = {.pTarget = pTable};
nodesWalkExpr(pTagCondCopy, rewriteAuthTable, &cxt);
if (NULL == *pWhere) { if (NULL == *pWhere) {
*pWhere = pTagCondCopy; *pWhere = pTagCondCopy;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -117,7 +135,7 @@ static EDealRes authSelectImpl(SNode* pNode, void* pContext) {
STableNode* pTable = (STableNode*)pNode; STableNode* pTable = (STableNode*)pNode;
pAuthCxt->errCode = checkAuth(pAuthCxt, pTable->dbName, pTable->tableName, AUTH_TYPE_READ, &pTagCond); pAuthCxt->errCode = checkAuth(pAuthCxt, pTable->dbName, pTable->tableName, AUTH_TYPE_READ, &pTagCond);
if (TSDB_CODE_SUCCESS == pAuthCxt->errCode && NULL != pTagCond) { if (TSDB_CODE_SUCCESS == pAuthCxt->errCode && NULL != pTagCond) {
pAuthCxt->errCode = appendStableTagCond(&pCxt->pSelect->pWhere, pTagCond); pAuthCxt->errCode = rewriteAppendStableTagCond(&pCxt->pSelect->pWhere, pTagCond, pTable);
} }
return TSDB_CODE_SUCCESS == pAuthCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR; return TSDB_CODE_SUCCESS == pAuthCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR;
} else if (QUERY_NODE_TEMP_TABLE == nodeType(pNode)) { } else if (QUERY_NODE_TEMP_TABLE == nodeType(pNode)) {
@ -152,7 +170,7 @@ static int32_t authDelete(SAuthCxt* pCxt, SDeleteStmt* pDelete) {
STableNode* pTable = (STableNode*)pDelete->pFromTable; STableNode* pTable = (STableNode*)pDelete->pFromTable;
int32_t code = checkAuth(pCxt, pTable->dbName, pTable->tableName, AUTH_TYPE_WRITE, &pTagCond); int32_t code = checkAuth(pCxt, pTable->dbName, pTable->tableName, AUTH_TYPE_WRITE, &pTagCond);
if (TSDB_CODE_SUCCESS == code && NULL != pTagCond) { if (TSDB_CODE_SUCCESS == code && NULL != pTagCond) {
code = appendStableTagCond(&pDelete->pWhere, pTagCond); code = rewriteAppendStableTagCond(&pDelete->pWhere, pTagCond, pTable);
} }
return code; return code;
} }

68
source/libs/stream/CMakeLists.txt
View File

@ -8,39 +8,43 @@ target_include_directories(
if(${BUILD_WITH_ROCKSDB}) if(${BUILD_WITH_ROCKSDB})
IF (TD_LINUX) if (${BUILD_CONTRIB})
target_link_libraries(
stream
PUBLIC rocksdb tdb
PRIVATE os util transport qcom executor wal index
)
target_include_directories( target_include_directories(
stream stream
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static" PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include"
) )
target_link_directories( else()
stream if (TD_LINUX)
PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static" target_include_directories(
) stream
target_link_libraries( PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static"
stream )
PUBLIC rocksdb tdb target_link_directories(
PRIVATE os util transport qcom executor wal index stream
) PUBLIC "${TD_SOURCE_DIR}/deps/${TD_DEPS_DIR}/rocksdb_static"
target_link_libraries( )
stream target_link_libraries(
PUBLIC rocksdb tdb stream
PRIVATE os util transport qcom executor wal index PUBLIC rocksdb tdb
) PRIVATE os util transport qcom executor wal index
ELSE() )
target_link_libraries( else()
stream target_link_libraries(
PUBLIC rocksdb tdb stream
PRIVATE os util transport qcom executor wal index PUBLIC rocksdb tdb
) PRIVATE os util transport qcom executor wal index
target_include_directories( )
stream target_include_directories(
PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include" stream
) PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include"
)
ENDIF() endif()
endif()
add_definitions(-DUSE_ROCKSDB) add_definitions(-DUSE_ROCKSDB)
endif(${BUILD_WITH_ROCKSDB}) endif(${BUILD_WITH_ROCKSDB})

2
source/libs/sync/src/syncMain.c
View File

@ -1590,6 +1590,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
// persist cfg // persist cfg
syncWriteCfgFile(pSyncNode); syncWriteCfgFile(pSyncNode);
#if 0
// change isStandBy to normal (election timeout) // change isStandBy to normal (election timeout)
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(pSyncNode, ""); syncNodeBecomeLeader(pSyncNode, "");
@ -1601,6 +1602,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
} else { } else {
syncNodeBecomeFollower(pSyncNode, ""); syncNodeBecomeFollower(pSyncNode, "");
} }
#endif
} else { } else {
// persist cfg // persist cfg
syncWriteCfgFile(pSyncNode); syncWriteCfgFile(pSyncNode);

6
source/util/src/tlosertree.c
View File

@ -71,12 +71,12 @@ int32_t tMergeTreeCreate(SMultiwayMergeTreeInfo** pTree, uint32_t numOfSources,
return 0; return 0;
} }
void tMergeTreeDestroy(SMultiwayMergeTreeInfo* pTree) { void tMergeTreeDestroy(SMultiwayMergeTreeInfo** pTree) {
if (pTree == NULL) { if (pTree == NULL || *pTree == NULL) {
return; return;
} }
taosMemoryFreeClear(pTree); taosMemoryFreeClear(*pTree);
} }
void tMergeTreeAdjust(SMultiwayMergeTreeInfo* pTree, int32_t idx) { void tMergeTreeAdjust(SMultiwayMergeTreeInfo* pTree, int32_t idx) {

56
tests/system-test/0-others/information_schema.py
View File

@ -132,8 +132,61 @@ class TDTestCase:
else: else:
tdSql.checkEqual(result[i][0],f'stb_{i-1}') tdSql.checkEqual(result[i][0],f'stb_{i-1}')
tdSql.checkEqual(result[i][1],ctbnum) tdSql.checkEqual(result[i][1],ctbnum)
def ins_stable_check2(self):
tdSql.execute('drop database if exists restful_test')
tdSql.execute('drop database if exists log')
tdSql.execute('drop database if exists d0')
tdSql.execute('drop database if exists d1')
tdSql.execute('create database restful_test vgroups 4 replica 1')
tdSql.execute('create database log vgroups 2 replica 1')
tdSql.execute('create database d0 vgroups 4 replica 1')
tdSql.execute('create database d1 vgroups 4 replica 1')
log_stb_num = 5
rest_stb_num = 51
for i in range(rest_stb_num):
tdSql.execute(f'create stable restful_test._stb_{i} (ts timestamp,c0 int) tags(t0 int);')
tdSql.execute(f'create stable d0._stb_{i} (ts timestamp,c0 int, c1 int) tags(t0 int,t1 int);')
tdSql.execute(f'create stable d1._stb_{i} (ts timestamp,c0 int, c1 int, c2 int) tags(t0 int,t1 int, t2 int);')
tdSql.execute(f'CREATE STABLE log.`taosadapter_restful_http_request_summary_milliseconds` (`_ts` TIMESTAMP, `sum` DOUBLE) TAGS (`request_uri` NCHAR(128));')
tdSql.execute(f'CREATE STABLE log.`taosadapter_system_cpu_percent` (`_ts` TIMESTAMP, `gauge` DOUBLE) TAGS (`endpoint` NCHAR(45));')
tdSql.execute(f'CREATE STABLE log.`taosadapter_restful_http_request_total` (`_ts` TIMESTAMP, `gauge` DOUBLE) TAGS (`client_ip` NCHAR(40));')
tdSql.execute(f'CREATE STABLE log.`taosadapter_system_mem_percent` (`_ts` TIMESTAMP, `gauge` DOUBLE) TAGS (`endpoint` NCHAR(45));')
tdSql.execute(f'CREATE STABLE log.`taosadapter_restful_http_request_fail` (`_ts` TIMESTAMP, `gauge` DOUBLE) TAGS (`request_uri` NCHAR(128), `status_code` NCHAR(4));')
tdSql.query(f'select * from information_schema.ins_stables where db_name="restful_test" limit 0,25;') # condition 1
result = tdSql.queryResult
tdSql.checkEqual(len(result),25)
for i in range(len(result)):
tdSql.checkEqual(result[i][0][0:5],f'_stb_') # stable_name
tdSql.checkEqual(result[i][1],f'restful_test') # db_name
tdSql.checkEqual(result[i][5]>=result[i][2],True) # last_update >= create_time
tdSql.checkEqual(result[i][3]>1,True) # columns
tdSql.checkEqual(result[i][4]>0,True) # tags
tdSql.checkEqual(result[i][6],None) # table_comment
tdSql.checkEqual(result[i][7],f'5000a,5000a') # watermark
tdSql.checkEqual(result[i][8],f'-1a,-1a') # max_delay
tdSql.checkEqual(result[i][9],f'') # rollup
tdSql.query(f'select create_time from information_schema.ins_stables where db_name="restful_test" order by create_time asc limit 10,1')
result = tdSql.queryResult
tdSql.checkEqual(len(result),1)
_create_time=result[0][0]
tdSql.query("select * from information_schema.ins_stables where db_name='restful_test' and create_time > '%s' limit 10,30" % (_create_time)) # condition 2
result = tdSql.queryResult
tdSql.checkEqual(len(result),30)
for i in range(len(result)):
tdSql.checkEqual(result[i][0][0:5],f'_stb_') # stable_name
tdSql.checkEqual(result[i][1],f'restful_test') # db_name
tdSql.checkEqual(result[i][5]>=result[i][2],True) # last_update >= create_time
tdSql.checkEqual(result[i][2]>_create_time,True) # create_time
tdSql.checkEqual(result[i][3]>1,True) # columns
tdSql.checkEqual(result[i][4]>0,True) # tags
tdSql.checkEqual(result[i][6],None) # table_comment
tdSql.checkEqual(result[i][7],f'5000a,5000a') # watermark
tdSql.checkEqual(result[i][8],f'-1a,-1a') # max_delay
tdSql.checkEqual(result[i][9],f'') # rollup
def ins_columns_check(self): def ins_columns_check(self):
tdSql.execute('drop database if exists db2') tdSql.execute('drop database if exists db2')
@ -216,6 +269,7 @@ class TDTestCase:
self.ins_columns_check() self.ins_columns_check()
# self.ins_col_check_4096() # self.ins_col_check_4096()
self.ins_stable_check() self.ins_stable_check()
self.ins_stable_check2()
self.ins_dnodes_check() self.ins_dnodes_check()
@ -224,4 +278,4 @@ class TDTestCase:
tdLog.success("%s successfully executed" % __file__) tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase()) tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase()) tdCases.addLinux(__file__, TDTestCase())

30
tools/CMakeLists.txt
View File

@ -136,7 +136,33 @@ ELSE ()
COMMAND cmake -E echo "Copy taosadapter-debug.exe" COMMAND cmake -E echo "Copy taosadapter-debug.exe"
COMMAND cmake -E copy taosadapter-debug.exe ${CMAKE_BINARY_DIR}/build/bin COMMAND cmake -E copy taosadapter-debug.exe ${CMAKE_BINARY_DIR}/build/bin
) )
ELSE (TD_WINDOWS) ELSEIF (TD_DARWIN)
MESSAGE("Building taosAdapter on MACOS")
INCLUDE(ExternalProject)
ExternalProject_Add(taosadapter
PREFIX "taosadapter"
SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/taosadapter
BUILD_ALWAYS off
DEPENDS taos
BUILD_IN_SOURCE 1
CONFIGURE_COMMAND cmake -E echo "taosadapter no need cmake to config"
PATCH_COMMAND
COMMAND git clean -f -d
BUILD_COMMAND
# COMMAND CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib go build -a -ldflags "-X github.com/taosdata/taosadapter/v3/version.Version=${taos_version} -X github.com/taosdata/taosadapter/v3/version.CommitID=${taosadapter_commit_sha1}"
COMMAND CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib go build -a -ldflags "-s -w -X github.com/taosdata/taosadapter/v3/version.Version=${taos_version} -X github.com/taosdata/taosadapter/v3/version.CommitID=${taosadapter_commit_sha1}"
COMMAND CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib go build -a -o taosadapter-debug -ldflags "-X github.com/taosdata/taosadapter/v3/version.Version=${taos_version} -X github.com/taosdata/taosadapter/v3/version.CommitID=${taosadapter_commit_sha1}"
INSTALL_COMMAND
COMMAND cmake -E echo "Copy taosadapter"
COMMAND cmake -E copy taosadapter ${CMAKE_BINARY_DIR}/build/bin
COMMAND cmake -E make_directory ${CMAKE_BINARY_DIR}/test/cfg/
COMMAND cmake -E echo "Copy taosadapter.toml"
COMMAND cmake -E copy ./example/config/taosadapter.toml ${CMAKE_BINARY_DIR}/test/cfg/
COMMAND cmake -E copy ./taosadapter.service ${CMAKE_BINARY_DIR}/test/cfg/
COMMAND cmake -E echo "Copy taosadapter-debug"
COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin
)
ELSE ()
MESSAGE("Building taosAdapter on non-Windows") MESSAGE("Building taosAdapter on non-Windows")
INCLUDE(ExternalProject) INCLUDE(ExternalProject)
ExternalProject_Add(taosadapter ExternalProject_Add(taosadapter
@ -164,5 +190,5 @@ ELSE ()
COMMAND cmake -E echo "Copy taosadapter-debug" COMMAND cmake -E echo "Copy taosadapter-debug"
COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin
) )
ENDIF (TD_WINDOWS) ENDIF ()
ENDIF () ENDIF ()