tq.h 6.4 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * 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/>.
 */
H
Hongze Cheng 已提交
15 16 17 18

#ifndef _TD_VNODE_TQ_H_
#define _TD_VNODE_TQ_H_

H
Hongze Cheng 已提交
19 20
#include "vnodeInt.h"

L
Liu Jicong 已提交
21 22 23 24
#include "executor.h"
#include "os.h"
#include "thash.h"
#include "tmsg.h"
25
#include "tqueue.h"
L
Liu Jicong 已提交
26
#include "trpc.h"
L
Liu Jicong 已提交
27 28 29
#include "ttimer.h"
#include "wal.h"

H
Hongze Cheng 已提交
30 31 32 33
#ifdef __cplusplus
extern "C" {
#endif

H
Hongze Cheng 已提交
34 35
// tqDebug ===================
// clang-format off
L
Liu Jicong 已提交
36 37 38 39 40 41
#define tqFatal(...) do { if (tqDebugFlag & DEBUG_FATAL) { taosPrintLog("TQ  FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
#define tqError(...) do { if (tqDebugFlag & DEBUG_ERROR) { taosPrintLog("TQ  ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}     while(0)
#define tqWarn(...)  do { if (tqDebugFlag & DEBUG_WARN)  { taosPrintLog("TQ  WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}       while(0)
#define tqInfo(...)  do { if (tqDebugFlag & DEBUG_INFO)  { taosPrintLog("TQ  ", DEBUG_INFO, 255, __VA_ARGS__); }}            while(0)
#define tqDebug(...) do { if (tqDebugFlag & DEBUG_DEBUG) { taosPrintLog("TQ  ", DEBUG_DEBUG, tqDebugFlag, __VA_ARGS__); }} while(0)
#define tqTrace(...) do { if (tqDebugFlag & DEBUG_TRACE) { taosPrintLog("TQ  ", DEBUG_TRACE, tqDebugFlag, __VA_ARGS__); }} while(0)
L
Liu Jicong 已提交
42

H
Hongze Cheng 已提交
43 44
// clang-format on

H
Hongze Cheng 已提交
45 46
typedef struct STqOffsetStore STqOffsetStore;

L
Liu Jicong 已提交
47 48
// tqPush

L
Liu Jicong 已提交
49
typedef struct {
L
Liu Jicong 已提交
50
  // msg info
L
Liu Jicong 已提交
51
  int64_t consumerId;
L
Liu Jicong 已提交
52 53
  int64_t reqOffset;
  int64_t processedVer;
L
Liu Jicong 已提交
54
  int32_t epoch;
L
Liu Jicong 已提交
55 56 57
  // rpc info
  int64_t        reqId;
  SRpcHandleInfo rpcInfo;
L
Liu Jicong 已提交
58 59
  tmr_h          timerId;
  int8_t         tmrStopped;
L
Liu Jicong 已提交
60
  // exec
L
Liu Jicong 已提交
61 62 63 64
  int8_t       inputStatus;
  int8_t       execStatus;
  SStreamQueue inputQ;
  SRWLatch     lock;
L
Liu Jicong 已提交
65
} STqPushHandle;
L
Liu Jicong 已提交
66 67 68 69

// tqExec

typedef struct {
70
  char* qmsg;   // SubPlanToString
L
Liu Jicong 已提交
71 72 73
} STqExecCol;

typedef struct {
L
Liu Jicong 已提交
74
  int64_t suid;
L
Liu Jicong 已提交
75 76 77
} STqExecTb;

typedef struct {
L
Liu Jicong 已提交
78
  SHashObj* pFilterOutTbUid;
L
Liu Jicong 已提交
79 80 81
} STqExecDb;

typedef struct {
82
  int8_t       subType;
83
  STqReader*   pTqReader;
84
  qTaskInfo_t  task;
L
Liu Jicong 已提交
85 86 87 88
  union {
    STqExecCol execCol;
    STqExecTb  execTb;
    STqExecDb  execDb;
L
Liu Jicong 已提交
89
  };
90
  int32_t      numOfCols;  // number of out pout column, temporarily used
L
Liu Jicong 已提交
91
} STqExecHandle;
H
Hongze Cheng 已提交
92

L
Liu Jicong 已提交
93
typedef struct {
94 95 96 97 98 99 100 101 102
  char          subKey[TSDB_SUBSCRIBE_KEY_LEN];
  int64_t       consumerId;
  int32_t       epoch;
  int8_t        fetchMeta;
  int64_t       snapshotVer;
  SWalReader*   pWalReader;
  SWalRef*      pRef;
  STqPushHandle pushHandle;    // push
  STqExecHandle execHandle;    // exec
L
Liu Jicong 已提交
103
} STqHandle;
104

L
Liu Jicong 已提交
105
typedef struct {
H
Haojun Liao 已提交
106
  SMqDataRsp*    pDataRsp;
L
Liu Jicong 已提交
107
  char           subKey[TSDB_SUBSCRIBE_KEY_LEN];
H
Haojun Liao 已提交
108
  SRpcHandleInfo info;
L
Liu Jicong 已提交
109 110
} STqPushEntry;

H
Hongze Cheng 已提交
111
struct STQ {
112 113 114 115 116 117 118
  SVnode*         pVnode;
  char*           path;
  int64_t         walLogLastVer;
  SRWLatch        lock;
  SHashObj*       pPushMgr;    // consumerId -> STqPushEntry
  SHashObj*       pHandle;     // subKey -> STqHandle
  SHashObj*       pCheckInfo;  // topic -> SAlterCheckInfo
119
  STqOffsetStore* pOffsetStore;
120 121 122 123
  TDB*            pMetaDB;
  TTB*            pExecStore;
  TTB*            pCheckStore;
  SStreamMeta*    pStreamMeta;
H
Hongze Cheng 已提交
124 125 126 127 128 129 130
};

typedef struct {
  int8_t inited;
  tmr_h  timer;
} STqMgmt;

131 132 133 134
typedef struct {
  int32_t size;
} STqOffsetHead;

L
Liu Jicong 已提交
135
static STqMgmt tqMgmt = {0};
H
Hongze Cheng 已提交
136

L
Liu Jicong 已提交
137 138 139
int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);

L
Liu Jicong 已提交
140
// tqRead
L
Liu Jicong 已提交
141
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
142
int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset);
143
int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum, uint64_t reqId);
L
Liu Jicong 已提交
144

wmmhello's avatar
wmmhello 已提交
145
// tqExec
146
int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp, int32_t* totalRows);
L
Liu Jicong 已提交
147
int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision);
H
Haojun Liao 已提交
148
int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp, int32_t type);
L
Liu Jicong 已提交
149
int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry);
L
Liu Jicong 已提交
150

L
Liu Jicong 已提交
151 152 153 154 155
// tqMeta
int32_t tqMetaOpen(STQ* pTq);
int32_t tqMetaClose(STQ* pTq);
int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
int32_t tqMetaDeleteHandle(STQ* pTq, const char* key);
L
Liu Jicong 已提交
156
int32_t tqMetaRestoreHandle(STQ* pTq);
157 158 159
int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_t vLen);
int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key);
int32_t tqMetaRestoreCheckInfo(STQ* pTq);
L
Liu Jicong 已提交
160

161
STqOffsetStore* tqOffsetOpen(STQ* pTq);
L
Liu Jicong 已提交
162
void            tqOffsetClose(STqOffsetStore*);
163 164
STqOffset*      tqOffsetRead(STqOffsetStore* pStore, const char* subscribeKey);
int32_t         tqOffsetWrite(STqOffsetStore* pStore, const STqOffset* pOffset);
L
Liu Jicong 已提交
165
int32_t         tqOffsetDelete(STqOffsetStore* pStore, const char* subscribeKey);
L
Liu Jicong 已提交
166
int32_t         tqOffsetCommitFile(STqOffsetStore* pStore);
167 168

// tqSink
L
Liu Jicong 已提交
169 170 171
int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBlock* pDataBlock,
                         SBatchDeleteReq* deleteReq);
void    tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
H
Hongze Cheng 已提交
172

L
Liu Jicong 已提交
173
// tqOffset
174
char*   tqOffsetBuildFName(const char* path, int32_t fVer);
L
Liu Jicong 已提交
175 176
int32_t tqOffsetRestoreFromFile(STqOffsetStore* pStore, const char* fname);

L
Liu Jicong 已提交
177
// tqStream
178
int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver);
179 180 181 182 183 184
int32_t tqDoRestoreSourceStreamTasks(STQ* pTq);

// tq util
void    createStreamTaskOffsetKey(char* dst, uint64_t streamId, uint32_t taskId);
int32_t tqAddInputBlockNLaunchTask(SStreamTask* pTask, SStreamQueueItem* pQueueItem, int64_t ver);
int32_t launchTaskForWalBlock(SStreamTask* pTask, SFetchRet* pRet, STqOffset* pOffset);
185
int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg);
L
Liu Jicong 已提交
186

187 188 189 190
void    doSaveTaskOffset(STqOffsetStore* pOffsetStore, const char* pKey, int64_t ver);
void    saveOffsetForAllTasks(STQ* pTq, int64_t ver);
void    initOffsetForAllRestoreTasks(STQ* pTq);

H
Hongze Cheng 已提交
191 192 193 194
#ifdef __cplusplus
}
#endif

L
Liu Jicong 已提交
195
#endif /*_TD_VNODE_TQ_H_*/