tqMeta.c 11.1 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/>.
 */
L
Liu Jicong 已提交
15 16 17
#include "tdbInt.h"
#include "tq.h"

L
Liu Jicong 已提交
18
int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
L
Liu Jicong 已提交
19 20
  if (tStartEncode(pEncoder) < 0) return -1;
  if (tEncodeCStr(pEncoder, pHandle->subKey) < 0) return -1;
21
  if (tEncodeI8(pEncoder, pHandle->fetchMeta) < 0) return -1;
L
Liu Jicong 已提交
22
  if (tEncodeI64(pEncoder, pHandle->consumerId) < 0) return -1;
L
Liu Jicong 已提交
23
  if (tEncodeI64(pEncoder, pHandle->snapshotVer) < 0) return -1;
L
Liu Jicong 已提交
24 25 26
  if (tEncodeI32(pEncoder, pHandle->epoch) < 0) return -1;
  if (tEncodeI8(pEncoder, pHandle->execHandle.subType) < 0) return -1;
  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
L
Liu Jicong 已提交
27
    if (tEncodeCStr(pEncoder, pHandle->execHandle.execCol.qmsg) < 0) return -1;
28
  } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
29 30
    int32_t size = taosHashGetSize(pHandle->execHandle.execDb.pFilterOutTbUid);
    if (tEncodeI32(pEncoder, size) < 0) return -1;
31
    void* pIter = NULL;
32
    pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter);
33 34
    while (pIter) {
      int64_t* tbUid = (int64_t*)taosHashGetKey(pIter, NULL);
35 36 37
      if (tEncodeI64(pEncoder, *tbUid) < 0) return -1;
      pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter);
    }
38
  } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
39
    if (tEncodeI64(pEncoder, pHandle->execHandle.execTb.suid) < 0) return -1;
L
Liu Jicong 已提交
40 41 42 43 44
  }
  tEndEncode(pEncoder);
  return pEncoder->pos;
}

L
Liu Jicong 已提交
45
int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
L
Liu Jicong 已提交
46 47
  if (tStartDecode(pDecoder) < 0) return -1;
  if (tDecodeCStrTo(pDecoder, pHandle->subKey) < 0) return -1;
48
  if (tDecodeI8(pDecoder, &pHandle->fetchMeta) < 0) return -1;
L
Liu Jicong 已提交
49
  if (tDecodeI64(pDecoder, &pHandle->consumerId) < 0) return -1;
L
Liu Jicong 已提交
50
  if (tDecodeI64(pDecoder, &pHandle->snapshotVer) < 0) return -1;
L
Liu Jicong 已提交
51 52 53
  if (tDecodeI32(pDecoder, &pHandle->epoch) < 0) return -1;
  if (tDecodeI8(pDecoder, &pHandle->execHandle.subType) < 0) return -1;
  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
L
Liu Jicong 已提交
54
    if (tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.execCol.qmsg) < 0) return -1;
55
  } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
56 57 58 59
    pHandle->execHandle.execDb.pFilterOutTbUid =
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
    int32_t size = 0;
    if (tDecodeI32(pDecoder, &size) < 0) return -1;
60
    for (int32_t i = 0; i < size; i++) {
61 62 63 64
      int64_t tbUid = 0;
      if (tDecodeI64(pDecoder, &tbUid) < 0) return -1;
      taosHashPut(pHandle->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
    }
65
  } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
66
    if (tDecodeI64(pDecoder, &pHandle->execHandle.execTb.suid) < 0) return -1;
L
Liu Jicong 已提交
67 68 69 70 71
  }
  tEndDecode(pDecoder);
  return 0;
}

72
int32_t tqMetaOpen(STQ* pTq) {
73
  if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaDB, 0) < 0) {
L
Liu Jicong 已提交
74
    return -1;
L
Liu Jicong 已提交
75 76
  }

77
  if (tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pExecStore, 0) < 0) {
78 79
    return -1;
  }
L
Liu Jicong 已提交
80

81
  if (tdbTbOpen("tq.check.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pCheckStore, 0) < 0) {
82 83
    return -1;
  }
L
Liu Jicong 已提交
84

85 86 87
  if (tqMetaRestoreHandle(pTq) < 0) {
    return -1;
  }
L
Liu Jicong 已提交
88

89 90 91
  if (tqMetaRestoreCheckInfo(pTq) < 0) {
    return -1;
  }
L
Liu Jicong 已提交
92

93 94
  return 0;
}
wmmhello's avatar
wmmhello 已提交
95

96 97 98
int32_t tqMetaClose(STQ* pTq) {
  if (pTq->pExecStore) {
    tdbTbClose(pTq->pExecStore);
L
Liu Jicong 已提交
99
  }
100 101 102 103
  if (pTq->pCheckStore) {
    tdbTbClose(pTq->pCheckStore);
  }
  tdbClose(pTq->pMetaDB);
L
Liu Jicong 已提交
104 105
  return 0;
}
L
Liu Jicong 已提交
106

107
int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_t vLen) {
108
  TXN* txn;
wmmhello's avatar
wmmhello 已提交
109

110 111
  if (tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
      0) {
L
Liu Jicong 已提交
112 113
    return -1;
  }
wmmhello's avatar
wmmhello 已提交
114

115
  if (tdbTbUpsert(pTq->pCheckStore, key, strlen(key), value, vLen, txn) < 0) {
116 117
    return -1;
  }
wmmhello's avatar
wmmhello 已提交
118

119
  if (tdbCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
120
    return -1;
L
Liu Jicong 已提交
121 122
  }

123
  if (tdbPostCommit(pTq->pMetaDB, txn) < 0) {
M
Minglei Jin 已提交
124 125 126
    return -1;
  }

L
Liu Jicong 已提交
127 128 129
  return 0;
}

130
int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key) {
131
  TXN* txn;
L
Liu Jicong 已提交
132

133 134
  if (tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
      0) {
L
Liu Jicong 已提交
135
    return -1;
L
Liu Jicong 已提交
136 137
  }

138
  if (tdbTbDelete(pTq->pCheckStore, key, (int)strlen(key), txn) < 0) {
L
Liu Jicong 已提交
139
    tqWarn("vgId:%d, tq try delete checkinfo failed %s", pTq->pVnode->config.vgId, key);
140 141
  }

142
  if (tdbCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
143
    return -1;
L
Liu Jicong 已提交
144 145
  }

146
  if (tdbPostCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
147
    return -1;
M
Minglei Jin 已提交
148 149
  }

L
Liu Jicong 已提交
150 151 152
  return 0;
}

153 154 155 156
int32_t tqMetaRestoreCheckInfo(STQ* pTq) {
  TBC* pCur = NULL;
  if (tdbTbcOpen(pTq->pCheckStore, &pCur, NULL) < 0) {
    return -1;
157
  }
158 159 160 161 162 163 164 165 166 167 168 169 170 171

  void*    pKey = NULL;
  int      kLen = 0;
  void*    pVal = NULL;
  int      vLen = 0;
  SDecoder decoder;

  tdbTbcMoveToFirst(pCur);

  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
    STqCheckInfo info;
    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
    if (tDecodeSTqCheckInfo(&decoder, &info) < 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
172
      tdbFree(pKey);
M
Minglei Jin 已提交
173
      tdbFree(pVal);
L
Liu Jicong 已提交
174
      tdbTbcClose(pCur);
175 176 177 178 179
      return -1;
    }
    tDecoderClear(&decoder);
    if (taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo)) < 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
180
      tdbFree(pKey);
M
Minglei Jin 已提交
181
      tdbFree(pVal);
L
Liu Jicong 已提交
182
      tdbTbcClose(pCur);
183 184
      return -1;
    }
185
  }
186
  tdbFree(pKey);
M
Minglei Jin 已提交
187
  tdbFree(pVal);
188
  tdbTbcClose(pCur);
L
Liu Jicong 已提交
189 190 191 192 193 194 195
  return 0;
}

int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
  int32_t code;
  int32_t vlen;
  tEncodeSize(tEncodeSTqHandle, pHandle, vlen, code);
L
Liu Jicong 已提交
196 197 198
  if (code < 0) {
    return -1;
  }
L
Liu Jicong 已提交
199

200
  tqDebug("tq save %s(%d) handle consumer:0x%" PRIx64 " epoch:%d vgId:%d", pHandle->subKey,
201
          (int32_t)strlen(pHandle->subKey), pHandle->consumerId, pHandle->epoch, TD_VID(pTq->pVnode));
M
Minglei Jin 已提交
202

L
Liu Jicong 已提交
203 204
  void* buf = taosMemoryCalloc(1, vlen);
  if (buf == NULL) {
L
Liu Jicong 已提交
205
    return -1;
L
Liu Jicong 已提交
206 207 208 209 210 211
  }

  SEncoder encoder;
  tEncoderInit(&encoder, buf, vlen);

  if (tEncodeSTqHandle(&encoder, pHandle) < 0) {
L
Liu Jicong 已提交
212 213 214
    tEncoderClear(&encoder);
    taosMemoryFree(buf);
    return -1;
L
Liu Jicong 已提交
215 216
  }

217
  TXN* txn;
L
Liu Jicong 已提交
218

219 220
  if (tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
      0) {
L
Liu Jicong 已提交
221 222 223
    tEncoderClear(&encoder);
    taosMemoryFree(buf);
    return -1;
L
Liu Jicong 已提交
224 225
  }

226
  if (tdbTbUpsert(pTq->pExecStore, key, (int)strlen(key), buf, vlen, txn) < 0) {
L
Liu Jicong 已提交
227 228 229
    tEncoderClear(&encoder);
    taosMemoryFree(buf);
    return -1;
L
Liu Jicong 已提交
230 231
  }

232
  if (tdbCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
233 234 235
    tEncoderClear(&encoder);
    taosMemoryFree(buf);
    return -1;
L
Liu Jicong 已提交
236 237
  }

238
  if (tdbPostCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
239 240 241
    tEncoderClear(&encoder);
    taosMemoryFree(buf);
    return -1;
M
Minglei Jin 已提交
242 243
  }

L
Liu Jicong 已提交
244 245 246 247 248 249
  tEncoderClear(&encoder);
  taosMemoryFree(buf);
  return 0;
}

int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
250
  TXN* txn;
L
Liu Jicong 已提交
251

252 253
  if (tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
      0) {
L
Liu Jicong 已提交
254
    return -1;
L
Liu Jicong 已提交
255 256
  }

257
  if (tdbTbDelete(pTq->pExecStore, key, (int)strlen(key), txn) < 0) {
L
Liu Jicong 已提交
258 259
  }

260
  if (tdbCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
261
    return -1;
L
Liu Jicong 已提交
262 263
  }

264
  if (tdbPostCommit(pTq->pMetaDB, txn) < 0) {
L
Liu Jicong 已提交
265
    return -1;
M
Minglei Jin 已提交
266 267
  }

L
Liu Jicong 已提交
268 269
  return 0;
}
270 271

int32_t tqMetaRestoreHandle(STQ* pTq) {
wmmhello's avatar
wmmhello 已提交
272
  int code = 0;
273 274 275 276 277
  TBC* pCur = NULL;
  if (tdbTbcOpen(pTq->pExecStore, &pCur, NULL) < 0) {
    return -1;
  }

D
dapan1121 已提交
278
  int32_t  vgId = TD_VID(pTq->pVnode);
279 280 281 282 283 284 285 286 287
  void*    pKey = NULL;
  int      kLen = 0;
  void*    pVal = NULL;
  int      vLen = 0;
  SDecoder decoder;

  tdbTbcMoveToFirst(pCur);

  while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
D
dapan1121 已提交
288
    STqHandle handle = {0};
289 290
    tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
    tDecodeSTqHandle(&decoder, &handle);
L
Liu Jicong 已提交
291
    tDecoderClear(&decoder);
292 293 294

    handle.pRef = walOpenRef(pTq->pVnode->pWal);
    if (handle.pRef == NULL) {
wmmhello's avatar
wmmhello 已提交
295 296
      code = -1;
      goto end;
297 298 299
    }
    walRefVer(handle.pRef, handle.snapshotVer);

wmmhello's avatar
wmmhello 已提交
300 301 302 303 304 305 306 307
    SReadHandle reader = {
        .meta = pTq->pVnode->pMeta,
        .vnode = pTq->pVnode,
        .initTableReader = true,
        .initTqReader = true,
        .version = handle.snapshotVer,
    };

308
    if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
L
Liu Jicong 已提交
309
      handle.execHandle.task =
D
dapan1121 已提交
310
          qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, vgId, &handle.execHandle.numOfCols, NULL);
L
Liu Jicong 已提交
311 312
      if (handle.execHandle.task == NULL) {
        tqError("cannot create exec task for %s", handle.subKey);
wmmhello's avatar
wmmhello 已提交
313 314
        code = -1;
        goto end;
L
Liu Jicong 已提交
315
      }
316
      void* scanner = NULL;
wmmhello's avatar
wmmhello 已提交
317
      qExtractStreamScanner(handle.execHandle.task, &scanner);
L
Liu Jicong 已提交
318 319
      if (scanner == NULL) {
        tqError("cannot extract stream scanner for %s", handle.subKey);
wmmhello's avatar
wmmhello 已提交
320 321
        code = -1;
        goto end;
L
Liu Jicong 已提交
322
      }
323
      handle.execHandle.pExecReader = qExtractReaderFromStreamScanner(scanner);
L
Liu Jicong 已提交
324 325
      if (handle.execHandle.pExecReader == NULL) {
        tqError("cannot extract exec reader for %s", handle.subKey);
wmmhello's avatar
wmmhello 已提交
326 327
        code = -1;
        goto end;
L
Liu Jicong 已提交
328
      }
329
    } else if (handle.execHandle.subType == TOPIC_SUB_TYPE__DB) {
wmmhello's avatar
wmmhello 已提交
330
      handle.pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
331 332
      handle.execHandle.pExecReader = tqOpenReader(pTq->pVnode);

333 334
      buildSnapContext(reader.meta, reader.version, 0, handle.execHandle.subType, handle.fetchMeta,
                       (SSnapContext**)(&reader.sContext));
D
dapan1121 已提交
335
      handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, NULL);
336 337
    } else if (handle.execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
      handle.pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
wmmhello's avatar
wmmhello 已提交
338

339 340 341 342 343 344 345 346 347 348 349
      SArray* tbUidList = taosArrayInit(0, sizeof(int64_t));
      vnodeGetCtbIdList(pTq->pVnode, handle.execHandle.execTb.suid, tbUidList);
      tqDebug("vgId:%d, tq try to get all ctb, suid:%" PRId64, pTq->pVnode->config.vgId, handle.execHandle.execTb.suid);
      for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
        int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
        tqDebug("vgId:%d, idx %d, uid:%" PRId64, TD_VID(pTq->pVnode), i, tbUid);
      }
      handle.execHandle.pExecReader = tqOpenReader(pTq->pVnode);
      tqReaderSetTbUidList(handle.execHandle.pExecReader, tbUidList);
      taosArrayDestroy(tbUidList);

350 351
      buildSnapContext(reader.meta, reader.version, handle.execHandle.execTb.suid, handle.execHandle.subType,
                       handle.fetchMeta, (SSnapContext**)(&reader.sContext));
D
dapan1121 已提交
352
      handle.execHandle.task = qCreateQueueExecTaskInfo(NULL, &reader, vgId, NULL, NULL);
353 354 355 356 357
    }
    tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle.subKey, handle.consumerId, TD_VID(pTq->pVnode));
    taosHashPut(pTq->pHandle, pKey, kLen, &handle, sizeof(STqHandle));
  }

wmmhello's avatar
wmmhello 已提交
358
end:
L
Liu Jicong 已提交
359 360
  tdbFree(pKey);
  tdbFree(pVal);
361
  tdbTbcClose(pCur);
wmmhello's avatar
wmmhello 已提交
362
  return code;
363
}