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

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/vnode_compact

...@@ -2,7 +2,7 @@ ...@@ -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 f0c1753 GIT_TAG 5662a6d
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
......
...@@ -30,6 +30,8 @@ database_option: { ...@@ -30,6 +30,8 @@ database_option: {
| WAL_LEVEL {1 | 2} | WAL_LEVEL {1 | 2}
| VGROUPS value | VGROUPS value
| SINGLE_STABLE {0 | 1} | SINGLE_STABLE {0 | 1}
| TABLE_PREFIX value
| TABLE_SUFFIX value
| WAL_RETENTION_PERIOD value | WAL_RETENTION_PERIOD value
| WAL_ROLL_PERIOD value | WAL_ROLL_PERIOD value
| WAL_RETENTION_SIZE value | WAL_RETENTION_SIZE value
...@@ -67,6 +69,8 @@ database_option: { ...@@ -67,6 +69,8 @@ database_option: {
- SINGLE_STABLE: specifies whether the database can contain more than one supertable. - SINGLE_STABLE: specifies whether the database can contain more than one supertable.
- 0: The database can contain multiple supertables. - 0: The database can contain multiple supertables.
- 1: The database can contain only one supertable. - 1: The database can contain only one supertable.
- TABLE_PREFIX:The prefix length in the table name that is ignored when distributing table to vnode based on table name.
- TABLE_SUFFIX:The suffix length in the table name that is ignored when distributing table to vnode based on table name.
- WAL_RETENTION_PERIOD: specifies the time after which WAL files are deleted. This parameter is used for data subscription. Enter a time in seconds. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is 4 days. - WAL_RETENTION_PERIOD: specifies the time after which WAL files are deleted. This parameter is used for data subscription. Enter a time in seconds. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is 4 days.
- WAL_RETENTION_SIZE: specifies the size at which WAL files are deleted. This parameter is used for data subscription. Enter a size in KB. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is -1. - WAL_RETENTION_SIZE: specifies the size at which WAL files are deleted. This parameter is used for data subscription. Enter a size in KB. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is -1.
- WAL_ROLL_PERIOD: specifies the time after which WAL files are rotated. After this period elapses, a new WAL file is created. The default value of single copy is 0. A value of 0 indicates that a new WAL file is created only after the previous WAL file was written to disk. The default values of multiple copy is 1 day. - WAL_ROLL_PERIOD: specifies the time after which WAL files are rotated. After this period elapses, a new WAL file is created. The default value of single copy is 0. A value of 0 indicates that a new WAL file is created only after the previous WAL file was written to disk. The default values of multiple copy is 1 day.
......
...@@ -30,6 +30,8 @@ database_option: { ...@@ -30,6 +30,8 @@ database_option: {
| WAL_LEVEL {1 | 2} | WAL_LEVEL {1 | 2}
| VGROUPS value | VGROUPS value
| SINGLE_STABLE {0 | 1} | SINGLE_STABLE {0 | 1}
| TABLE_PREFIX value
| TABLE_SUFFIX value
| WAL_RETENTION_PERIOD value | WAL_RETENTION_PERIOD value
| WAL_ROLL_PERIOD value | WAL_ROLL_PERIOD value
| WAL_RETENTION_SIZE value | WAL_RETENTION_SIZE value
...@@ -67,6 +69,8 @@ database_option: { ...@@ -67,6 +69,8 @@ database_option: {
- SINGLE_STABLE:表示此数据库中是否只可以创建一个超级表,用于超级表列非常多的情况。 - SINGLE_STABLE:表示此数据库中是否只可以创建一个超级表,用于超级表列非常多的情况。
- 0:表示可以创建多张超级表。 - 0:表示可以创建多张超级表。
- 1:表示只可以创建一张超级表。 - 1:表示只可以创建一张超级表。
- TABLE_PREFIX:内部存储引擎根据表名分配存储该表数据的 VNODE 时要忽略的前缀的长度。
- TABLE_SUFFIX:内部存储引擎根据表名分配存储该表数据的 VNODE 时要忽略的后缀的长度。
- WAL_RETENTION_PERIOD:wal 文件的额外保留策略,用于数据订阅。wal 的保存时长,单位为 s。单副本默认为 0,即落盘后立即删除。-1 表示不删除。多副本默认为 4 天。 - WAL_RETENTION_PERIOD:wal 文件的额外保留策略,用于数据订阅。wal 的保存时长,单位为 s。单副本默认为 0,即落盘后立即删除。-1 表示不删除。多副本默认为 4 天。
- WAL_RETENTION_SIZE:wal 文件的额外保留策略,用于数据订阅。wal 的保存的最大上限,单位为 KB。单副本默认为 0,即落盘后立即删除。多副本默认为-1,表示不删除。 - WAL_RETENTION_SIZE:wal 文件的额外保留策略,用于数据订阅。wal 的保存的最大上限,单位为 KB。单副本默认为 0,即落盘后立即删除。多副本默认为-1,表示不删除。
- WAL_ROLL_PERIOD:wal 文件切换时长,单位为 s。当 wal 文件创建并写入后,经过该时间,会自动创建一个新的 wal 文件。单副本默认为 0,即仅在落盘时创建新文件。多副本默认为 1 天。 - WAL_ROLL_PERIOD:wal 文件切换时长,单位为 s。当 wal 文件创建并写入后,经过该时间,会自动创建一个新的 wal 文件。单副本默认为 0,即仅在落盘时创建新文件。多副本默认为 1 天。
......
...@@ -213,6 +213,7 @@ int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver); ...@@ -213,6 +213,7 @@ int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver);
int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver); int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver);
int32_t qStreamRecoverFinish(qTaskInfo_t tinfo); int32_t qStreamRecoverFinish(qTaskInfo_t tinfo);
int32_t qStreamRestoreParam(qTaskInfo_t tinfo); int32_t qStreamRestoreParam(qTaskInfo_t tinfo);
bool qStreamRecoverScanFinished(qTaskInfo_t tinfo);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -19,7 +19,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) { ...@@ -19,7 +19,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {
memset(pReader, 0, sizeof(*pReader)); memset(pReader, 0, sizeof(*pReader));
pReader->flags = flags; pReader->flags = flags;
pReader->pMeta = pMeta; pReader->pMeta = pMeta;
if (!(flags & META_READER_NOLOCK)) { if (pReader->pMeta && !(flags & META_READER_NOLOCK)) {
metaRLock(pMeta); metaRLock(pMeta);
} }
} }
...@@ -152,7 +152,7 @@ bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid) { ...@@ -152,7 +152,7 @@ bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid) {
} }
int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) { int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
SMeta *pMeta = pReader->pMeta; SMeta *pMeta = pReader->pMeta;
int64_t version1; int64_t version1;
// query uid.idx // query uid.idx
...@@ -239,7 +239,6 @@ int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName) { ...@@ -239,7 +239,6 @@ int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName) {
return 0; return 0;
} }
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) { int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) {
int code = 0; int code = 0;
SMetaReader mr = {0}; SMetaReader mr = {0};
...@@ -756,9 +755,7 @@ int64_t metaGetTimeSeriesNum(SMeta *pMeta) { ...@@ -756,9 +755,7 @@ int64_t metaGetTimeSeriesNum(SMeta *pMeta) {
return pMeta->pVnode->config.vndStats.numOfTimeSeries + pMeta->pVnode->config.vndStats.numOfNTimeSeries; return pMeta->pVnode->config.vndStats.numOfTimeSeries + pMeta->pVnode->config.vndStats.numOfNTimeSeries;
} }
int64_t metaGetNtbNum(SMeta *pMeta) { int64_t metaGetNtbNum(SMeta *pMeta) { return pMeta->pVnode->config.vndStats.numOfNTables; }
return pMeta->pVnode->config.vndStats.numOfNTables;
}
typedef struct { typedef struct {
SMeta *pMeta; SMeta *pMeta;
......
...@@ -136,6 +136,7 @@ typedef struct { ...@@ -136,6 +136,7 @@ typedef struct {
SSchemaWrapper* schema; SSchemaWrapper* schema;
char tbName[TSDB_TABLE_NAME_LEN]; char tbName[TSDB_TABLE_NAME_LEN];
int8_t recoverStep; int8_t recoverStep;
int8_t recoverScanFinished;
SQueryTableDataCond tableCond; SQueryTableDataCond tableCond;
int64_t fillHistoryVer1; int64_t fillHistoryVer1;
int64_t fillHistoryVer2; int64_t fillHistoryVer2;
...@@ -182,7 +183,7 @@ struct SExecTaskInfo { ...@@ -182,7 +183,7 @@ struct SExecTaskInfo {
SSubplan* pSubplan; SSubplan* pSubplan;
struct SOperatorInfo* pRoot; struct SOperatorInfo* pRoot;
SLocalFetch localFetch; SLocalFetch localFetch;
SArray* pResultBlockList;// result block list SArray* pResultBlockList; // result block list
STaskStopInfo stopInfo; STaskStopInfo stopInfo;
}; };
...@@ -199,7 +200,7 @@ typedef struct SOperatorFpSet { ...@@ -199,7 +200,7 @@ typedef struct SOperatorFpSet {
__optr_fn_t getNextFn; __optr_fn_t getNextFn;
__optr_fn_t cleanupFn; // call this function to release the allocated resources ASAP __optr_fn_t cleanupFn; // call this function to release the allocated resources ASAP
__optr_close_fn_t closeFn; __optr_close_fn_t closeFn;
__optr_reqBuf_fn_t reqBufFn; // total used buffer for blocking operator __optr_reqBuf_fn_t reqBufFn; // total used buffer for blocking operator
__optr_encode_fn_t encodeResultRow; __optr_encode_fn_t encodeResultRow;
__optr_decode_fn_t decodeResultRow; __optr_decode_fn_t decodeResultRow;
__optr_explain_fn_t getExplainFn; __optr_explain_fn_t getExplainFn;
...@@ -255,22 +256,22 @@ typedef struct SLimitInfo { ...@@ -255,22 +256,22 @@ typedef struct SLimitInfo {
} SLimitInfo; } SLimitInfo;
typedef struct SExchangeInfo { typedef struct SExchangeInfo {
SArray* pSources; SArray* pSources;
SArray* pSourceDataInfo; SArray* pSourceDataInfo;
tsem_t ready; tsem_t ready;
void* pTransporter; void* pTransporter;
// SArray<SSDataBlock*>, result block list, used to keep the multi-block that // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
// passed by downstream operator // passed by downstream operator
SArray* pResultBlockList; SArray* pResultBlockList;
SArray* pRecycledBlocks;// build a pool for small data block to avoid to repeatly create and then destroy. SArray* pRecycledBlocks; // build a pool for small data block to avoid to repeatly create and then destroy.
SSDataBlock* pDummyBlock; // dummy block, not keep data SSDataBlock* pDummyBlock; // dummy block, not keep data
bool seqLoadData; // sequential load data or not, false by default bool seqLoadData; // sequential load data or not, false by default
int32_t current; int32_t current;
SLoadRemoteDataInfo loadInfo; SLoadRemoteDataInfo loadInfo;
uint64_t self; uint64_t self;
SLimitInfo limitInfo; SLimitInfo limitInfo;
int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo
} SExchangeInfo; } SExchangeInfo;
typedef struct SScanInfo { typedef struct SScanInfo {
...@@ -305,9 +306,9 @@ typedef struct { ...@@ -305,9 +306,9 @@ typedef struct {
} SAggOptrPushDownInfo; } SAggOptrPushDownInfo;
typedef struct STableMetaCacheInfo { typedef struct STableMetaCacheInfo {
SLRUCache* pTableMetaEntryCache; // 100 by default SLRUCache* pTableMetaEntryCache; // 100 by default
uint64_t metaFetch; uint64_t metaFetch;
uint64_t cacheHit; uint64_t cacheHit;
} STableMetaCacheInfo; } STableMetaCacheInfo;
typedef struct STableScanBase { typedef struct STableScanBase {
...@@ -325,16 +326,16 @@ typedef struct STableScanBase { ...@@ -325,16 +326,16 @@ typedef struct STableScanBase {
} STableScanBase; } STableScanBase;
typedef struct STableScanInfo { typedef struct STableScanInfo {
STableScanBase base; STableScanBase base;
SScanInfo scanInfo; SScanInfo scanInfo;
int32_t scanTimes; int32_t scanTimes;
SSDataBlock* pResBlock; SSDataBlock* pResBlock;
SSampleExecInfo sample; // sample execution info SSampleExecInfo sample; // sample execution info
int32_t currentGroupId; int32_t currentGroupId;
int32_t currentTable; int32_t currentTable;
int8_t scanMode; int8_t scanMode;
int8_t assignBlockUid; int8_t assignBlockUid;
bool hasGroupByTag; bool hasGroupByTag;
} STableScanInfo; } STableScanInfo;
typedef struct STableMergeScanInfo { typedef struct STableMergeScanInfo {
...@@ -360,11 +361,11 @@ typedef struct STableMergeScanInfo { ...@@ -360,11 +361,11 @@ typedef struct STableMergeScanInfo {
} STableMergeScanInfo; } STableMergeScanInfo;
typedef struct STagScanInfo { typedef struct STagScanInfo {
SColumnInfo* pCols; SColumnInfo* pCols;
SSDataBlock* pRes; SSDataBlock* pRes;
SColMatchInfo matchInfo; SColMatchInfo matchInfo;
int32_t curPos; int32_t curPos;
SReadHandle readHandle; SReadHandle readHandle;
} STagScanInfo; } STagScanInfo;
typedef enum EStreamScanMode { typedef enum EStreamScanMode {
...@@ -468,6 +469,11 @@ typedef struct SStreamScanInfo { ...@@ -468,6 +469,11 @@ typedef struct SStreamScanInfo {
SNodeList* pGroupTags; SNodeList* pGroupTags;
SNode* pTagCond; SNode* pTagCond;
SNode* pTagIndexCond; SNode* pTagIndexCond;
// recover
int32_t blockRecoverContiCnt;
int32_t blockRecoverTotCnt;
} SStreamScanInfo; } SStreamScanInfo;
typedef struct { typedef struct {
...@@ -499,8 +505,8 @@ typedef struct STableCountScanOperatorInfo { ...@@ -499,8 +505,8 @@ typedef struct STableCountScanOperatorInfo {
STableCountScanSupp supp; STableCountScanSupp supp;
int32_t currGrpIdx; int32_t currGrpIdx;
SArray* stbUidList; // when group by db_name and/or stable_name SArray* stbUidList; // when group by db_name and/or stable_name
} STableCountScanOperatorInfo; } STableCountScanOperatorInfo;
typedef struct SOptrBasicInfo { typedef struct SOptrBasicInfo {
...@@ -678,19 +684,19 @@ void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32 ...@@ -678,19 +684,19 @@ void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32
void destroyOperatorInfo(SOperatorInfo* pOperator); void destroyOperatorInfo(SOperatorInfo* pOperator);
int32_t optrDefaultBufFn(SOperatorInfo* pOperator); int32_t optrDefaultBufFn(SOperatorInfo* pOperator);
void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock); void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void cleanupBasicInfo(SOptrBasicInfo* pInfo); void cleanupBasicInfo(SOptrBasicInfo* pInfo);
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr); int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
void cleanupExprSupp(SExprSupp* pSup); void cleanupExprSupp(SExprSupp* pSup);
void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs); void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize, int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
const char* pkey); const char* pkey);
void cleanupAggSup(SAggSupporter* pAggSup); void cleanupAggSup(SAggSupporter* pAggSup);
void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows); void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
SDiskbasedBuf* pBuf); SDiskbasedBuf* pBuf);
...@@ -803,10 +809,10 @@ void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order, ...@@ -803,10 +809,10 @@ void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order,
int32_t checkForQueryBuf(size_t numOfTables); int32_t checkForQueryBuf(size_t numOfTables);
bool isTaskKilled(SExecTaskInfo* pTaskInfo); bool isTaskKilled(SExecTaskInfo* pTaskInfo);
void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode); void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
void doDestroyTask(SExecTaskInfo* pTaskInfo); void doDestroyTask(SExecTaskInfo* pTaskInfo);
void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
char* sql, EOPTR_EXEC_MODEL model); char* sql, EOPTR_EXEC_MODEL model);
...@@ -828,8 +834,8 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup); ...@@ -828,8 +834,8 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup); bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
uint64_t* pGp, void* pTbName); uint64_t* pGp, void* pTbName);
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock); void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup, int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
......
...@@ -937,6 +937,11 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) { ...@@ -937,6 +937,11 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) {
return 0; return 0;
} }
bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
return pTaskInfo->streamInfo.recoverScanFinished;
}
void* qExtractReaderFromStreamScanner(void* scanner) { void* qExtractReaderFromStreamScanner(void* scanner) {
SStreamScanInfo* pInfo = scanner; SStreamScanInfo* pInfo = scanner;
return (void*)pInfo->tqReader; return (void*)pInfo->tqReader;
......
...@@ -1785,11 +1785,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { ...@@ -1785,11 +1785,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pTSInfo->scanTimes = 0; pTSInfo->scanTimes = 0;
pTSInfo->currentGroupId = -1; pTSInfo->currentGroupId = -1;
pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN; pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN;
pTaskInfo->streamInfo.recoverScanFinished = false;
} }
if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) { if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) {
if (pInfo->blockRecoverContiCnt > 100) {
pInfo->blockRecoverTotCnt += pInfo->blockRecoverContiCnt;
pInfo->blockRecoverContiCnt = 0;
return NULL;
}
SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp);
if (pBlock != NULL) { if (pBlock != NULL) {
pInfo->blockRecoverContiCnt++;
calBlockTbName(pInfo, pBlock); calBlockTbName(pInfo, pBlock);
if (pInfo->pUpdateInfo) { if (pInfo->pUpdateInfo) {
TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex);
...@@ -1807,6 +1814,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { ...@@ -1807,6 +1814,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pTSInfo->base.cond.startVersion = -1; pTSInfo->base.cond.startVersion = -1;
pTSInfo->base.cond.endVersion = -1; pTSInfo->base.cond.endVersion = -1;
pTaskInfo->streamInfo.recoverScanFinished = true;
return NULL; return NULL;
} }
......
...@@ -52,8 +52,8 @@ TEST_F(ParserExplainToSyncdbTest, grant) { ...@@ -52,8 +52,8 @@ TEST_F(ParserExplainToSyncdbTest, grant) {
ASSERT_EQ(string(req.objname), string(expect.objname)); ASSERT_EQ(string(req.objname), string(expect.objname));
}); });
setAlterUserReq(TSDB_ALTER_USER_ADD_ALL_DB, "wxy", "0.test"); setAlterUserReq(TSDB_ALTER_USER_ADD_ALL_DB, "wxy", "0.*");
run("GRANT ALL ON test.* TO wxy"); run("GRANT ALL ON *.* TO wxy");
setAlterUserReq(TSDB_ALTER_USER_ADD_READ_DB, "wxy", "0.test"); setAlterUserReq(TSDB_ALTER_USER_ADD_READ_DB, "wxy", "0.test");
run("GRANT READ ON test.* TO wxy"); run("GRANT READ ON test.* TO wxy");
...@@ -138,10 +138,38 @@ TEST_F(ParserExplainToSyncdbTest, redistributeVgroup) { ...@@ -138,10 +138,38 @@ TEST_F(ParserExplainToSyncdbTest, redistributeVgroup) {
TEST_F(ParserExplainToSyncdbTest, revoke) { TEST_F(ParserExplainToSyncdbTest, revoke) {
useDb("root", "test"); useDb("root", "test");
run("REVOKE ALL ON test.* FROM wxy"); SAlterUserReq expect = {0};
auto setAlterUserReq = [&](int8_t alterType, const string& user, const string& obj) {
expect.alterType = alterType;
snprintf(expect.user, sizeof(expect.user), "%s", user.c_str());
snprintf(expect.objname, sizeof(expect.objname), "%s", obj.c_str());
};
setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_REVOKE_STMT);
ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_ALTER_USER);
SAlterUserReq req = {0};
ASSERT_EQ(tDeserializeSAlterUserReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
ASSERT_EQ(req.alterType, expect.alterType);
ASSERT_EQ(string(req.user), string(expect.user));
ASSERT_EQ(string(req.objname), string(expect.objname));
});
setAlterUserReq(TSDB_ALTER_USER_REMOVE_ALL_DB, "wxy", "0.*");
run("REVOKE ALL ON *.* FROM wxy");
setAlterUserReq(TSDB_ALTER_USER_REMOVE_READ_DB, "wxy", "0.test");
run("REVOKE READ ON test.* FROM wxy"); run("REVOKE READ ON test.* FROM wxy");
setAlterUserReq(TSDB_ALTER_USER_REMOVE_WRITE_DB, "wxy", "0.test");
run("REVOKE WRITE ON test.* FROM wxy"); run("REVOKE WRITE ON test.* FROM wxy");
setAlterUserReq(TSDB_ALTER_USER_REMOVE_ALL_DB, "wxy", "0.test");
run("REVOKE READ, WRITE ON test.* FROM wxy"); run("REVOKE READ, WRITE ON test.* FROM wxy");
setAlterUserReq(TSDB_ALTER_USER_REMOVE_SUBSCRIBE_TOPIC, "wxy", "0.tp1");
run("REVOKE SUBSCRIBE ON tp1 FROM wxy");
} }
// todo syncdb // todo syncdb
......
...@@ -99,7 +99,7 @@ TEST_F(ParserInitialDTest, dropDnode) { ...@@ -99,7 +99,7 @@ TEST_F(ParserInitialDTest, dropDnode) {
expect.force = force; expect.force = force;
}; };
auto setDropDnodeReqByEndpoint = [&](const char* pFqdn, int32_t port, bool force = false) { auto setDropDnodeReqByEndpoint = [&](const char* pFqdn, int32_t port = tsServerPort, bool force = false) {
strcpy(expect.fqdn, pFqdn); strcpy(expect.fqdn, pFqdn);
expect.port = port; expect.port = port;
expect.force = force; expect.force = force;
...@@ -131,6 +131,14 @@ TEST_F(ParserInitialDTest, dropDnode) { ...@@ -131,6 +131,14 @@ TEST_F(ParserInitialDTest, dropDnode) {
setDropDnodeReqByEndpoint("host2", 8030, true); setDropDnodeReqByEndpoint("host2", 8030, true);
run("DROP DNODE 'host2:8030' FORCE"); run("DROP DNODE 'host2:8030' FORCE");
clearDropDnodeReq(); clearDropDnodeReq();
setDropDnodeReqByEndpoint("host1");
run("DROP DNODE host1");
clearDropDnodeReq();
setDropDnodeReqByEndpoint("host2", tsServerPort, true);
run("DROP DNODE host2 FORCE");
clearDropDnodeReq();
} }
// todo DROP function // todo DROP function
...@@ -174,7 +182,21 @@ TEST_F(ParserInitialDTest, dropMnode) { ...@@ -174,7 +182,21 @@ TEST_F(ParserInitialDTest, dropMnode) {
TEST_F(ParserInitialDTest, dropQnode) { TEST_F(ParserInitialDTest, dropQnode) {
useDb("root", "test"); useDb("root", "test");
run("DROP qnode on dnode 1"); SMDropQnodeReq expect = {0};
auto setDropQnodeReq = [&](int32_t dnodeId) { expect.dnodeId = dnodeId; };
setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_DROP_QNODE_STMT);
SMDropQnodeReq req = {0};
ASSERT_TRUE(TSDB_CODE_SUCCESS ==
tDeserializeSCreateDropMQSNodeReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req));
ASSERT_EQ(req.dnodeId, expect.dnodeId);
});
setDropQnodeReq(1);
run("DROP QNODE ON DNODE 1");
} }
TEST_F(ParserInitialDTest, dropSnode) { TEST_F(ParserInitialDTest, dropSnode) {
...@@ -237,7 +259,20 @@ TEST_F(ParserInitialDTest, dropUser) { ...@@ -237,7 +259,20 @@ TEST_F(ParserInitialDTest, dropUser) {
login("root"); login("root");
useDb("root", "test"); useDb("root", "test");
run("DROP user wxy"); SDropUserReq expect = {0};
auto setDropUserReq = [&](const char* pUser) { sprintf(expect.user, "%s", pUser); };
setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_DROP_USER_STMT);
SDropUserReq req = {0};
ASSERT_TRUE(TSDB_CODE_SUCCESS == tDeserializeSDropUserReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req));
ASSERT_EQ(std::string(req.user), std::string(expect.user));
});
setDropUserReq("wxy");
run("DROP USER wxy");
} }
} // namespace ParserTest } // namespace ParserTest
...@@ -112,7 +112,11 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) { ...@@ -112,7 +112,11 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
ASSERT(0); ASSERT(0);
} }
if (output == NULL) { if (output == NULL) {
finished = true; if (qStreamRecoverScanFinished(exec)) {
finished = true;
} else {
qSetStreamOpOpen(exec);
}
break; break;
} }
......
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#define ALLOW_FORBID_FUNC #define ALLOW_FORBID_FUNC
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include <stdlib.h>
#ifdef WINDOWS #ifdef WINDOWS
void swapStr(char* j, char* J, int width) { void swapStr(char* j, char* J, int width) {
...@@ -33,16 +34,5 @@ void swapStr(char* j, char* J, int width) { ...@@ -33,16 +34,5 @@ void swapStr(char* j, char* J, int width) {
// todo refactor: 1) move away; 2) use merge sort instead; 3) qsort is not a stable sort actually. // todo refactor: 1) move away; 2) use merge sort instead; 3) qsort is not a stable sort actually.
void taosSort(void* arr, int64_t sz, int64_t width, __compar_fn_t compar) { void taosSort(void* arr, int64_t sz, int64_t width, __compar_fn_t compar) {
#ifdef WINDOWS
int64_t i, j;
for (i = 0; i < sz - 1; i++) {
for (j = 0; j < sz - 1 - i; j++) {
if (compar((char*)arr + j * width, (char*)arr + (j + 1) * width) > 0.00) {
swapStr((char*)arr + j * width, (char*)arr + (j + 1) * width, width);
}
}
}
#else
qsort(arr, sz, width, compar); qsort(arr, sz, width, compar);
#endif
} }
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册