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

Merge pull request #16836 from taosdata/feature/stream 

feat(stream): stream state support tuple
This commit is contained in:
Liu Jicong 2022-09-14 17:47:47 +08:00 committed by GitHub
parent 5589f78b32 c7f986a656
commit 3bedc3fa3b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 254 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
include/common/tcommon.h
View File

@ -45,8 +45,8 @@ enum {
// clang-format on
typedef struct {
TSKEY ts;
uint64_t groupId;
TSKEY ts;
} SWinKey;
static inline int SWinKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
@ -68,6 +68,37 @@ static inline int SWinKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, i
return 0;
}
typedef struct {
uint64_t groupId;
TSKEY ts;
int32_t exprIdx;
} STupleKey;
static inline int STupleKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
STupleKey* pTuple1 = (STupleKey*)pKey1;
STupleKey* pTuple2 = (STupleKey*)pKey2;
if (pTuple1->groupId > pTuple2->groupId) {
return 1;
} else if (pTuple1->groupId < pTuple2->groupId) {
return -1;
}
if (pTuple1->ts > pTuple2->ts) {
return 1;
} else if (pTuple1->ts < pTuple2->ts) {
return -1;
}
if (pTuple1->exprIdx > pTuple2->exprIdx) {
return 1;
} else if (pTuple1->exprIdx < pTuple2->exprIdx) {
return -1;
}
return 0;
}
enum {
TMQ_MSG_TYPE__DUMMY = 0,
TMQ_MSG_TYPE__POLL_RSP,

145
include/libs/function/function.h
View File

@ -34,66 +34,69 @@ typedef struct SFuncExecEnv {
int32_t calcMemSize;
} SFuncExecEnv;
typedef bool (*FExecGetEnv)(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
typedef bool (*FExecGetEnv)(struct SFunctionNode *pFunc, SFuncExecEnv *pEnv);
typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo *pResultCellInfo);
typedef int32_t (*FExecProcess)(struct SqlFunctionCtx *pCtx);
typedef int32_t (*FExecFinalize)(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock);
typedef int32_t (*FExecFinalize)(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock);
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef int32_t (*FExecCombine)(struct SqlFunctionCtx *pDestCtx, struct SqlFunctionCtx *pSourceCtx);
typedef struct SScalarFuncExecFuncs {
FExecGetEnv getEnv;
FExecGetEnv getEnv;
FScalarExecProcess process;
} SScalarFuncExecFuncs;
typedef struct SFuncExecFuncs {
FExecGetEnv getEnv;
FExecInit init;
FExecProcess process;
FExecGetEnv getEnv;
FExecInit init;
FExecProcess process;
FExecFinalize finalize;
FExecCombine combine;
FExecCombine combine;
} SFuncExecFuncs;
#define MAX_INTERVAL_TIME_WINDOW 1000000 // maximum allowed time windows in final results
#define MAX_INTERVAL_TIME_WINDOW 1000000 // maximum allowed time windows in final results
#define TOP_BOTTOM_QUERY_LIMIT 100
#define FUNCTIONS_NAME_MAX_LENGTH 16
typedef struct SResultRowEntryInfo {
bool initialized:1; // output buffer has been initialized
bool complete:1; // query has completed
uint8_t isNullRes:6; // the result is null
uint16_t numOfRes; // num of output result in current buffer. NOT NULL RESULT
bool initialized : 1; // output buffer has been initialized
bool complete : 1; // query has completed
uint8_t isNullRes : 6; // the result is null
uint16_t numOfRes; // num of output result in current buffer. NOT NULL RESULT
} SResultRowEntryInfo;
// determine the real data need to calculated the result
enum {
BLK_DATA_NOT_LOAD = 0x0,
BLK_DATA_SMA_LOAD = 0x1,
BLK_DATA_NOT_LOAD = 0x0,
BLK_DATA_SMA_LOAD = 0x1,
BLK_DATA_DATA_LOAD = 0x3,
BLK_DATA_FILTEROUT = 0x4, // discard current data block since it is not qualified for filter
BLK_DATA_FILTEROUT = 0x4, // discard current data block since it is not qualified for filter
};
enum {
MAIN_SCAN = 0x0u,
REVERSE_SCAN = 0x1u, // todo remove it
REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan
MERGE_STAGE = 0x20u,
MAIN_SCAN = 0x0u,
REVERSE_SCAN = 0x1u, // todo remove it
REPEAT_SCAN = 0x2u, // repeat scan belongs to the master scan
MERGE_STAGE = 0x20u,
};
typedef struct SPoint1 {
int64_t key;
union{double val; char* ptr;};
int64_t key;
union {
double val;
char *ptr;
};
} SPoint1;
struct SqlFunctionCtx;
struct SResultRowEntryInfo;
//for selectivity query, the corresponding tag value is assigned if the data is qualified
// for selectivity query, the corresponding tag value is assigned if the data is qualified
typedef struct SSubsidiaryResInfo {
int16_t num;
int32_t rowLen;
char* buf; // serialize data buffer
int16_t num;
int32_t rowLen;
char *buf; // serialize data buffer
struct SqlFunctionCtx **pCtx;
} SSubsidiaryResInfo;
@ -106,69 +109,70 @@ typedef struct SResultDataInfo {
} SResultDataInfo;
#define GET_RES_INFO(ctx) ((ctx)->resultInfo)
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowEntryInfo)))
#define GET_ROWCELL_INTERBUF(_c) ((void *)((char *)(_c) + sizeof(SResultRowEntryInfo)))
typedef struct SInputColumnInfoData {
int32_t totalRows; // total rows in current columnar data
int32_t startRowIndex; // handle started row index
int32_t numOfRows; // the number of rows needs to be handled
int32_t numOfInputCols; // PTS is not included
bool colDataAggIsSet;// if agg is set or not
SColumnInfoData *pPTS; // primary timestamp column
int32_t totalRows; // total rows in current columnar data
int32_t startRowIndex; // handle started row index
int32_t numOfRows; // the number of rows needs to be handled
int32_t numOfInputCols; // PTS is not included
bool colDataAggIsSet; // if agg is set or not
SColumnInfoData *pPTS; // primary timestamp column
SColumnInfoData **pData;
SColumnDataAgg **pColumnDataAgg;
uint64_t uid; // table uid, used to set the tag value when building the final query result for selectivity functions.
uint64_t uid; // table uid, used to set the tag value when building the final query result for selectivity functions.
} SInputColumnInfoData;
typedef struct SSerializeDataHandle {
struct SDiskbasedBuf* pBuf;
struct SDiskbasedBuf *pBuf;
int32_t currentPage;
void *pState;
} SSerializeDataHandle;
// sql function runtime context
typedef struct SqlFunctionCtx {
SInputColumnInfoData input;
SResultDataInfo resDataInfo;
uint32_t order; // data block scanner order: asc|desc
uint8_t scanFlag; // record current running step, default: 0
int16_t functionId; // function id
char *pOutput; // final result output buffer, point to sdata->data
int32_t numOfParams;
SFunctParam *param; // input parameter, e.g., top(k, 20), the number of results for top query is kept in param
SColumnInfoData *pTsOutput; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
int32_t offset;
struct SResultRowEntryInfo *resultInfo;
SSubsidiaryResInfo subsidiaries;
SPoint1 start;
SPoint1 end;
SFuncExecFuncs fpSet;
SScalarFuncExecFuncs sfp;
struct SExprInfo *pExpr;
struct SSDataBlock *pSrcBlock;
struct SSDataBlock *pDstBlock; // used by indefinite rows function to set selectivity
SSerializeDataHandle saveHandle;
bool isStream;
SInputColumnInfoData input;
SResultDataInfo resDataInfo;
uint32_t order; // data block scanner order: asc|desc
uint8_t scanFlag; // record current running step, default: 0
int16_t functionId; // function id
char *pOutput; // final result output buffer, point to sdata->data
int32_t numOfParams;
SFunctParam *param; // input parameter, e.g., top(k, 20), the number of results for top query is kept in param
SColumnInfoData *pTsOutput; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
int32_t offset;
struct SResultRowEntryInfo *resultInfo;
SSubsidiaryResInfo subsidiaries;
SPoint1 start;
SPoint1 end;
SFuncExecFuncs fpSet;
SScalarFuncExecFuncs sfp;
struct SExprInfo *pExpr;
struct SSDataBlock *pSrcBlock;
struct SSDataBlock *pDstBlock; // used by indefinite rows function to set selectivity
SSerializeDataHandle saveHandle;
bool isStream;
char udfName[TSDB_FUNC_NAME_LEN];
char udfName[TSDB_FUNC_NAME_LEN];
} SqlFunctionCtx;
enum {
TEXPR_BINARYEXPR_NODE= 0x1,
TEXPR_BINARYEXPR_NODE = 0x1,
TEXPR_UNARYEXPR_NODE = 0x2,
};
typedef struct tExprNode {
int32_t nodeType;
union {
struct {// function node
char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor
int32_t functionId;
int32_t num;
struct SFunctionNode *pFunctNode;
struct { // function node
char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor
int32_t functionId;
int32_t num;
struct SFunctionNode *pFunctNode;
} _function;
struct {
struct SNode* pRootNode;
struct SNode *pRootNode;
} _optrRoot;
};
} tExprNode;
@ -182,17 +186,18 @@ struct SScalarParam {
int32_t numOfRows;
};
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock);
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
void cleanupResultRowEntry(struct SResultRowEntryInfo *pCell);
int32_t getNumOfResult(SqlFunctionCtx *pCtx, int32_t num, SSDataBlock *pResBlock);
bool isRowEntryCompleted(struct SResultRowEntryInfo *pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo *pEntry);
typedef struct SPoint {
int64_t key;
void * val;
void *val;
} SPoint;
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2, int32_t inputType);
int32_t taosGetLinearInterpolationVal(SPoint *point, int32_t outputType, SPoint *point1, SPoint *point2,
int32_t inputType);
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// udf api

78
include/libs/stream/streamState.h Normal file
View File

@ -0,0 +1,78 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tdatablock.h"
#include "tdbInt.h"
#ifdef __cplusplus
extern "C" {
#endif
#ifndef _STREAM_STATE_H_
#define _STREAM_STATE_H_
typedef struct SStreamTask SStreamTask;
// incremental state storage
typedef struct {
SStreamTask* pOwner;
TDB* db;
TTB* pStateDb;
TTB* pFuncStateDb;
TXN txn;
} SStreamState;
SStreamState* streamStateOpen(char* path, SStreamTask* pTask);
void streamStateClose(SStreamState* pState);
int32_t streamStateBegin(SStreamState* pState);
int32_t streamStateCommit(SStreamState* pState);
int32_t streamStateAbort(SStreamState* pState);
typedef struct {
TBC* pCur;
} SStreamStateCur;
#if 1
int32_t streamStateFuncPut(SStreamState* pState, const STupleKey* key, const void* value, int32_t vLen);
int32_t streamStateFuncGet(SStreamState* pState, const STupleKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key);
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateReleaseBuf(SStreamState* pState, const SWinKey* key, void* pVal);
void streamFreeVal(void* val);
SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const SWinKey* key);
void streamStateFreeCur(SStreamStateCur* pCur);
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
#endif
#ifdef __cplusplus
}
#endif
#endif /* ifndef _STREAM_STATE_H_ */

42
include/libs/stream/tstream.h
View File

@ -16,6 +16,7 @@
#include "executor.h"
#include "os.h"
#include "query.h"
#include "streamState.h"
#include "tdatablock.h"
#include "tdbInt.h"
#include "tmsg.h"
@ -263,14 +264,6 @@ typedef struct {
SArray* checkpointVer;
} SStreamRecoveringState;
// incremental state storage
typedef struct {
SStreamTask* pOwner;
TDB* db;
TTB* pStateDb;
TXN txn;
} SStreamState;
typedef struct SStreamTask {
int64_t streamId;
int32_t taskId;
@ -540,39 +533,6 @@ int32_t streamMetaCommit(SStreamMeta* pMeta);
int32_t streamMetaRollBack(SStreamMeta* pMeta);
int32_t streamLoadTasks(SStreamMeta* pMeta);
SStreamState* streamStateOpen(char* path, SStreamTask* pTask);
void streamStateClose(SStreamState* pState);
int32_t streamStateBegin(SStreamState* pState);
int32_t streamStateCommit(SStreamState* pState);
int32_t streamStateAbort(SStreamState* pState);
typedef struct {
TBC* pCur;
} SStreamStateCur;
#if 1
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateReleaseBuf(SStreamState* pState, const SWinKey* key, void* pVal);
void streamFreeVal(void* val);
SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const SWinKey* key);
void streamStateFreeCur(SStreamStateCur* pCur);
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
#endif
#ifdef __cplusplus
}
#endif

1
include/util/taoserror.h
View File

@ -285,6 +285,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_TOPIC_SUBSCRIBED TAOS_DEF_ERROR_CODE(0, 0x03EB)
#define TSDB_CODE_MND_CGROUP_USED TAOS_DEF_ERROR_CODE(0, 0x03EC)
#define TSDB_CODE_MND_TOPIC_MUST_BE_DELETED TAOS_DEF_ERROR_CODE(0, 0x03ED)
#define TSDB_CODE_MND_IN_REBALANCE TAOS_DEF_ERROR_CODE(0, 0x03EF)
// mnode-stream
#define TSDB_CODE_MND_STREAM_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03F0)

1
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -900,6 +900,7 @@ int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName)
// iter all vnode to delete handle
if (taosHashGetSize(pSub->consumerHash) != 0) {
sdbRelease(pSdb, pSub);
terrno = TSDB_CODE_MND_IN_REBALANCE;
return -1;
}
int32_t sz = taosArrayGetSize(pSub->unassignedVgs);

1
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -713,7 +713,6 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
mndReleaseTopic(pMnode, pTopic);
if (code != 0) {
terrno = code;
mError("topic:%s, failed to drop since %s", dropReq.name, terrstr());
return -1;
}

2
source/libs/function/CMakeLists.txt
View File

@ -14,7 +14,7 @@ target_include_directories(
target_link_libraries(
function
PRIVATE os util common nodes scalar qcom transport
PRIVATE os util common nodes scalar qcom transport stream
PUBLIC uv_a
)

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

@ -18,6 +18,7 @@
#include "function.h"
#include "query.h"
#include "querynodes.h"
#include "streamState.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "tdigest.h"
@ -56,8 +57,13 @@ typedef struct SAvgRes {
} SAvgRes;
typedef struct STuplePos {
int32_t pageId;
int32_t offset;
union {
struct {
int32_t pageId;
int32_t offset;
};
STupleKey streamTupleKey;
};
} STuplePos;
typedef struct SMinmaxResInfo {
@ -1146,7 +1152,8 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
return true;
}
static STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock);
static STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock,
const STupleKey* pKey);
static int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
static const char* loadTupleData(SqlFunctionCtx* pCtx, const STuplePos* pPos);
@ -1201,7 +1208,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
pBuf->v = *(int64_t*)tval;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
} else {
if (IS_SIGNED_NUMERIC_TYPE(type)) {
@ -1213,7 +1220,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
*(int64_t*)&pBuf->v = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
}
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
@ -1225,7 +1232,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
*(uint64_t*)&pBuf->v = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
}
} else if (type == TSDB_DATA_TYPE_DOUBLE) {
@ -1237,7 +1244,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
*(double*)&pBuf->v = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
}
} else if (type == TSDB_DATA_TYPE_FLOAT) {
@ -1251,7 +1258,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
}
}
@ -1276,7 +1283,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1307,7 +1314,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1338,7 +1345,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1369,7 +1376,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1402,7 +1409,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1433,7 +1440,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1464,7 +1471,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1495,7 +1502,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1527,7 +1534,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1558,7 +1565,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
if (!pBuf->assign) {
*val = pData[i];
if (pCtx->subsidiaries.num > 0) {
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock, NULL);
}
pBuf->assign = true;
} else {
@ -1581,7 +1588,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
_min_max_over:
if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved) {
pBuf->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
pBuf->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, NULL);
pBuf->nullTupleSaved = true;
}
return numOfElems;
@ -2758,7 +2765,7 @@ static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowInde
}
if (!pInfo->hasResult) {
pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock);
pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock, NULL);
} else {
updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
}
@ -3426,7 +3433,7 @@ int32_t topFunction(SqlFunctionCtx* pCtx) {
}
if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, NULL);
pRes->nullTupleSaved = true;
}
return TSDB_CODE_SUCCESS;
@ -3454,7 +3461,7 @@ int32_t bottomFunction(SqlFunctionCtx* pCtx) {
}
if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, NULL);
pRes->nullTupleSaved = true;
}
@ -3506,7 +3513,7 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
// save the data of this tuple
if (pCtx->subsidiaries.num > 0) {
pItem->tuplePos = saveTupleData(pCtx, rowIndex, pSrcBlock);
pItem->tuplePos = saveTupleData(pCtx, rowIndex, pSrcBlock, NULL);
}
#ifdef BUF_PAGE_DEBUG
qDebug("page_saveTuple i:%d, item:%p,pageId:%d, offset:%d\n", pEntryInfo->numOfRes, pItem, pItem->tuplePos.pageId,
@ -3578,7 +3585,8 @@ void* serializeTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SSubsid
return buf;
}
static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length) {
static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length,
const STupleKey* pKey) {
STuplePos p = {0};
if (pHandle->pBuf != NULL) {
SFilePage* pPage = NULL;
@ -3604,12 +3612,16 @@ static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf
releaseBufPage(pHandle->pBuf, pPage);
} else {
// other tuple save policy
if (streamStateFuncPut(pHandle->pState, pKey, pBuf, length) < 0) {
ASSERT(0);
}
p.streamTupleKey = *pKey;
}
return p;
}
STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock) {
STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, const STupleKey* pKey) {
if (pCtx->subsidiaries.rowLen == 0) {
int32_t rowLen = 0;
for (int32_t j = 0; j < pCtx->subsidiaries.num; ++j) {
@ -3622,7 +3634,7 @@ STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBloc
}
char* buf = serializeTupleData(pSrcBlock, rowIndex, &pCtx->subsidiaries, pCtx->subsidiaries.buf);
return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen);
return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, pKey);
}
static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STuplePos* pPos) {
@ -3632,6 +3644,7 @@ static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf
setBufPageDirty(pPage, true);
releaseBufPage(pHandle->pBuf, pPage);
} else {
streamStateFuncPut(pHandle->pState, &pPos->streamTupleKey, pBuf, length);
}
return TSDB_CODE_SUCCESS;
@ -3650,7 +3663,10 @@ static char* doLoadTupleData(SSerializeDataHandle* pHandle, const STuplePos* pPo
releaseBufPage(pHandle->pBuf, pPage);
return p;
} else {
return NULL;
void* value = NULL;
int32_t vLen;
streamStateFuncGet(pHandle->pState, &pPos->streamTupleKey, &value, &vLen);
return (char*)value;
}
}
@ -4981,7 +4997,7 @@ static void doReservoirSample(SqlFunctionCtx* pCtx, SSampleInfo* pInfo, char* da
if (pInfo->numSampled < pInfo->samples) {
sampleAssignResult(pInfo, data, pInfo->numSampled);
if (pCtx->subsidiaries.num > 0) {
pInfo->tuplePos[pInfo->numSampled] = saveTupleData(pCtx, index, pCtx->pSrcBlock);
pInfo->tuplePos[pInfo->numSampled] = saveTupleData(pCtx, index, pCtx->pSrcBlock, NULL);
}
pInfo->numSampled++;
} else {
@ -5012,7 +5028,7 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
}
if (pInfo->numSampled == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
pInfo->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
pInfo->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, NULL);
pInfo->nullTupleSaved = true;
}

21
source/libs/stream/src/streamState.c
View File

@ -35,6 +35,10 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask) {
goto _err;
}
if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb) < 0) {
goto _err;
}
if (streamStateBegin(pState) < 0) {
goto _err;
}
@ -44,8 +48,9 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask) {
return pState;
_err:
if (pState->pStateDb) tdbTbClose(pState->pStateDb);
if (pState->db) tdbClose(pState->db);
tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb);
tdbClose(pState->db);
taosMemoryFree(pState);
return NULL;
}
@ -53,6 +58,7 @@ _err:
void streamStateClose(SStreamState* pState) {
tdbCommit(pState->db, &pState->txn);
tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb);
tdbClose(pState->db);
taosMemoryFree(pState);
@ -101,6 +107,17 @@ int32_t streamStateAbort(SStreamState* pState) {
return 0;
}
int32_t streamStateFuncPut(SStreamState* pState, const STupleKey* key, const void* value, int32_t vLen) {
return tdbTbUpsert(pState->pFuncStateDb, key, sizeof(STupleKey), value, vLen, &pState->txn);
}
int32_t streamStateFuncGet(SStreamState* pState, const STupleKey* key, void** pVal, int32_t* pVLen) {
return tdbTbGet(pState->pFuncStateDb, key, sizeof(STupleKey), pVal, pVLen);
}
int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key) {
return tdbTbDelete(pState->pFuncStateDb, key, sizeof(STupleKey), &pState->txn);
}
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
return tdbTbUpsert(pState->pStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
}

1
source/util/src/terror.c
View File

@ -288,6 +288,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_SUBSCRIBED, "Topic subscribed cannot be dropped")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_MUST_BE_DELETED, "Topic must be dropped first")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CGROUP_USED, "Consumer group being used by some consumer")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_IN_REBALANCE, "Topic being rebalanced")
// mnode-stream
TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_ALREADY_EXIST, "Stream already exists")