taosdata/TDengine | Build 4881 | include/libs/executor/storageapi.h | Coveralls

1

/*

2

 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>

3

4

 * This program is free software: you can use, redistribute, and/or modify

5

 * it under the terms of the GNU Affero General Public License, version 3

6

 * or later ("AGPL"), as published by the Free Software Foundation.

7

8

 * This program is distributed in the hope that it will be useful, but WITHOUT

9

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

10

 * FITNESS FOR A PARTICULAR PURPOSE.

11

12

 * You should have received a copy of the GNU Affero General Public License

13

 * along with this program. If not, see <http://www.gnu.org/licenses/>.

14

*/

15

16

#ifndef TDENGINE_STORAGEAPI_H

17

#define TDENGINE_STORAGEAPI_H

18

19

#include "function.h"

20

#include "index.h"

21

#include "osMemory.h"

22

#include "taosdef.h"

23

#include "tcommon.h"

24

#include "tmsg.h"

25

#include "tscalablebf.h"

26

#include "tsimplehash.h"

27

28

#ifdef __cplusplus

29

extern "C" {

30

#endif

31

32

#define TIMEWINDOW_RANGE_CONTAINED 1

33

#define TIMEWINDOW_RANGE_EXTERNAL  2

34

35

#define CACHESCAN_RETRIEVE_TYPE_ALL    0x1

36

#define CACHESCAN_RETRIEVE_TYPE_SINGLE 0x2

37

#define CACHESCAN_RETRIEVE_LAST_ROW    0x4

38

#define CACHESCAN_RETRIEVE_LAST        0x8

39

40

#define META_READER_LOCK   0x0

41

#define META_READER_NOLOCK 0x1

42

43

#define STREAM_STATE_BUFF_HASH        1

44

#define STREAM_STATE_BUFF_SORT        2

45

#define STREAM_STATE_BUFF_HASH_SORT   3

46

#define STREAM_STATE_BUFF_HASH_SEARCH 4

47

48

typedef struct SMeta SMeta;

49

typedef TSKEY (*GetTsFun)(void*);

50

51

typedef struct SMetaEntry {

52

  int64_t  version;

53

  int8_t   type;

54

  int8_t   flags;  // TODO: need refactor?

55

  tb_uid_t uid;

56

  char*    name;

57

  union {

58

    struct {

59

      SSchemaWrapper schemaRow;

60

      SSchemaWrapper schemaTag;

61

      SRSmaParam     rsmaParam;

62

      int64_t        keep;

63

    } stbEntry;

64

    struct {

65

      int64_t  btime;

66

      int32_t  ttlDays;

67

      int32_t  commentLen;  // not include '\0'

68

      char*    comment;

69

      tb_uid_t suid;

70

      uint8_t* pTags;

71

    } ctbEntry;

72

    struct {

73

      int64_t        btime;

74

      int32_t        ttlDays;

75

      int32_t        commentLen;

76

      char*          comment;

77

      int32_t        ncid;  // next column id

78

      SSchemaWrapper schemaRow;

79

    } ntbEntry;

80

    struct {

81

      STSma* tsma;

82

    } smaEntry;

83

};

84

85

  uint8_t* pBuf;

86

87

  SColCmprWrapper colCmpr;  // col compress alg

88

  SExtSchema*     pExtSchemas;

89

  SColRefWrapper  colRef;   // col reference for virtual table

90

} SMetaEntry;

91

92

typedef struct SMetaReader {

93

  int32_t            flags;

94

  void*              pMeta;

95

  SDecoder           coder;

96

  SMetaEntry         me;

97

  void*              pBuf;

98

  int32_t            szBuf;

99

  struct SStoreMeta* pAPI;

100

} SMetaReader;

101

102

typedef struct SMTbCursor {

103

  void*       pMeta;

104

  void*       pDbc;

105

  void*       pKey;

106

  void*       pVal;

107

  int32_t     kLen;

108

  int32_t     vLen;

109

  SMetaReader mr;

110

  int8_t      paused;

111

} SMTbCursor;

112

113

typedef struct SMCtbCursor {

114

  struct SMeta* pMeta;

115

  void*         pCur;

116

  tb_uid_t      suid;

117

  void*         pKey;

118

  void*         pVal;

119

  int           kLen;

120

  int           vLen;

121

  int8_t        paused;

122

  int           lock;

123

} SMCtbCursor;

124

125

typedef struct SRowBuffPos {

126

  void* pRowBuff;

127

  void* pKey;

128

  bool  beFlushed;

129

  bool  beUsed;

130

  bool  needFree;

131

  bool  beUpdated;

132

  bool  invalid;

133

} SRowBuffPos;

134

135

// tq

136

typedef struct SMetaTableInfo {

137

  int64_t         suid;

138

  int64_t         uid;

139

  SSchemaWrapper* schema;

140

  SExtSchema*     pExtSchemas;

141

  char            tbName[TSDB_TABLE_NAME_LEN];

142

} SMetaTableInfo;

143

144

static FORCE_INLINE void destroyMetaTableInfo(SMetaTableInfo* mtInfo){

148

149

150

typedef struct SSnapContext {

151

  struct SMeta* pMeta;

152

  int64_t       snapVersion;

153

  void*         pCur;

154

  int64_t       suid;

155

  int8_t        subType;

156

  SHashObj*     idVersion;

157

  SHashObj*     suidInfo;

158

  SArray*       idList;

159

  int32_t       index;

160

  int8_t        withMeta;

161

  int8_t        queryMeta;  // true-get meta, false-get data

162

  bool          hasPrimaryKey;

163

} SSnapContext;

164

165

typedef struct {

166

  int64_t uid;

167

  int64_t ctbNum;

168

  int32_t colNum;

169

  int8_t  flags;

170

  int64_t keep;

171

} SMetaStbStats;

172

173

// clang-format off

174

/*-------------------------------------------------new api format---------------------------------------------------*/

175

typedef enum {

176

  TSD_READER_NOTIFY_DURATION_START,

177

  TSD_READER_NOTIFY_NEXT_DURATION_BLOCK,

178

} ETsdReaderNotifyType;

179

180

typedef union {

181

  struct {

182

    int32_t filesetId;

183

  } duration;

184

} STsdReaderNotifyInfo;

185

186

typedef void (*TsdReaderNotifyCbFn)(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param);

187

188

struct SFileSetReader;

189

190

typedef struct TsdReader {

191

  int32_t      (*tsdReaderOpen)(void* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables,

192

                           SSDataBlock* pResBlock, void** ppReader, const char* idstr, SHashObj** pIgnoreTables);

193

  void         (*tsdReaderClose)(void* pReader);

194

  int32_t      (*tsdSetReaderTaskId)(void* pReader, const char* pId);

195

  int32_t      (*tsdSetQueryTableList)(void* p, const void* pTableList, int32_t num);

196

  int32_t      (*tsdNextDataBlock)(void* pReader, bool* hasNext);

197

198

  int32_t      (*tsdReaderRetrieveBlockSMAInfo)();

199

  int32_t      (*tsdReaderRetrieveDataBlock)(void* p, SSDataBlock** pBlock);

200

201

  void         (*tsdReaderReleaseDataBlock)(void* pReader);

202

203

  int32_t      (*tsdReaderResetStatus)(void* p, SQueryTableDataCond* pCond);

204

  int32_t      (*tsdReaderGetDataBlockDistInfo)();

205

  void         (*tsdReaderGetDatablock)();

206

  void         (*tsdReaderSetDatablock)();

207

  int64_t      (*tsdReaderGetNumOfInMemRows)();

208

  void         (*tsdReaderNotifyClosing)();

209

210

  void         (*tsdSetFilesetDelimited)(void* pReader);

211

  void         (*tsdSetSetNotifyCb)(void* pReader, TsdReaderNotifyCbFn notifyFn, void* param);

212

213

  // for fileset query

214

  int32_t (*fileSetReaderOpen)(void *pVnode, struct SFileSetReader **ppReader);

215

  int32_t (*fileSetReadNext)(struct SFileSetReader *);

216

  int32_t (*fileSetGetEntryField)(struct SFileSetReader *, const char *, void *);

217

  void (*fileSetReaderClose)(struct SFileSetReader **);

218

219

  // retrieve first/last ts for each table

220

  int32_t  (*tsdCreateFirstLastTsIter)(void *pVnode, STimeWindow *pWindow, SVersionRange *pVerRange, uint64_t suid, void *pTableList,

221

                                   int32_t numOfTables, int32_t order, void **pIter, const char *idstr);

222

  int32_t  (*tsdNextFirstLastTsBlock)(void *pIter, SSDataBlock *pRes, bool* hasNext);

223

  void     (*tsdDestroyFirstLastTsIter)(void *pIter);

224

225

} TsdReader;

226

227

typedef struct SStoreCacheReader {

228

  int32_t  (*openReader)(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,

229

                         SArray *pCidList, int32_t *pSlotIds, uint64_t suid, void **pReader, const char *idstr,

230

                         SArray *pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks);

231

  void     (*closeReader)(void *pReader);

232

  int32_t  (*retrieveRows)(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, const int32_t *dstSlotIds,

233

                           SArray *pTableUidList, bool* pGotAllRows);

234

  int32_t  (*reuseReader)(void *pReader, void *pTableIdList, int32_t numOfTables);

235

} SStoreCacheReader;

236

237

// clang-format on

238

239

/*------------------------------------------------------------------------------------------------------------------*/

240

// todo rename

241

typedef struct SStoreTqReader {

242

  struct STqReader* (*tqReaderOpen)();

243

  void (*tqReaderClose)();

244

245

  int32_t (*tqReaderSeek)();

246

  bool (*tqReaderNextBlockInWal)();

247

  bool (*tqNextBlockImpl)();  // todo remove it

248

  SSDataBlock* (*tqGetResultBlock)();

249

  int64_t (*tqGetResultBlockTime)();

250

251

  int32_t (*tqReaderSetColIdList)();

252

  int32_t (*tqReaderSetQueryTableList)();

253

254

  void (*tqReaderAddTables)();

255

  void (*tqReaderRemoveTables)();

256

257

  void (*tqSetTablePrimaryKey)();

258

  bool (*tqGetTablePrimaryKey)();

259

  bool (*tqReaderIsQueriedTable)();

260

  bool (*tqReaderCurrentBlockConsumed)();

261

262

  struct SWalReader* (*tqReaderGetWalReader)();  // todo remove it

263

                                                 //  int32_t (*tqReaderRetrieveTaosXBlock)();       // todo remove it

264

265

  int32_t (*tqReaderSetSubmitMsg)();  // todo remove it

266

  //  bool (*tqReaderNextBlockFilterOut)();

267

} SStoreTqReader;

268

269

typedef struct SStoreSnapshotFn {

270

  bool (*taosXGetTablePrimaryKey)(SSnapContext* ctx);

271

  void (*taosXSetTablePrimaryKey)(SSnapContext* ctx, int64_t uid);

272

  int32_t (*setForSnapShot)(SSnapContext* ctx, int64_t uid);

273

  void (*destroySnapshot)(SSnapContext* ctx);

274

  int32_t (*getMetaTableInfoFromSnapshot)(SSnapContext* ctx, SMetaTableInfo* info);

275

  int32_t (*getTableInfoFromSnapshot)(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid);

276

} SStoreSnapshotFn;

277

278

typedef struct SStoreMeta {

279

  SMTbCursor* (*openTableMetaCursor)(void* pVnode);                                 // metaOpenTbCursor

280

  void (*closeTableMetaCursor)(SMTbCursor* pTbCur);                                 // metaCloseTbCursor

281

  void (*pauseTableMetaCursor)(SMTbCursor* pTbCur);                                 // metaPauseTbCursor

282

  int32_t (*resumeTableMetaCursor)(SMTbCursor* pTbCur, int8_t first, int8_t move);  // metaResumeTbCursor

283

  int32_t (*cursorNext)(SMTbCursor* pTbCur, ETableType jumpTableType);              // metaTbCursorNext

284

  int32_t (*cursorPrev)(SMTbCursor* pTbCur, ETableType jumpTableType);              // metaTbCursorPrev

285

286

  int32_t (*getTableTags)(void* pVnode, uint64_t suid, SArray* uidList);

287

  int32_t (*getTableTagsByUid)(void* pVnode, int64_t suid, SArray* uidList);

288

  const void* (*extractTagVal)(const void* tag, int16_t type, STagVal* tagVal);  // todo remove it

289

290

  int32_t (*getTableUidByName)(void* pVnode, char* tbName, uint64_t* uid);

291

  int32_t (*getTableTypeSuidByName)(void* pVnode, char* tbName, ETableType* tbType, uint64_t* suid);

292

  int32_t (*getTableNameByUid)(void* pVnode, uint64_t uid, char* tbName);

293

  bool (*isTableExisted)(void* pVnode, tb_uid_t uid);

294

295

  int32_t (*metaGetCachedTbGroup)(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList);

296

  int32_t (*metaPutTbGroupToCache)(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,

297

                                   int32_t payloadLen);

298

299

  int32_t (*getCachedTableList)(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray* pList1,

300

                                bool* acquireRes);

301

  int32_t (*putCachedTableList)(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,

302

                                int32_t payloadLen, double selectivityRatio);

303

  int32_t (*getStableCachedTableList)(void* pVnode, tb_uid_t suid,

304

    const uint8_t* pTagCondKey, int32_t tagCondKeyLen,

305

    const uint8_t* pKey, int32_t keyLen, SArray* pList1, bool* acquireRes);

306

  int32_t (*putStableCachedTableList)(void* pVnode, uint64_t suid,

307

    const void* pTagCondKey, int32_t tagCondKeyLen,

308

    const void* pKey, int32_t keyLen, SArray* pUidList, SArray** pTagColIds);

309

310

  int32_t (*metaGetCachedRefDbs)(void* pVnode, tb_uid_t suid, SArray* pList);

311

  int32_t (*metaPutRefDbsToCache)(void* pVnode, tb_uid_t suid, SArray* pList);

312

313

  void* (*storeGetIndexInfo)(void* pVnode);

314

  void* (*getInvertIndex)(void* pVnode);

315

  // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]

316

  int32_t (*getChildTableList)(void* pVnode, int64_t suid, SArray* list);

317

  int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList);

318

  int32_t (*getTableSchema)(void* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid, SSchemaWrapper** pTagSchema);

319

  int32_t (*getNumOfChildTables)(void* pVnode, int64_t uid, int64_t* numOfTables, int32_t* numOfCols, int8_t* flags);

320

  void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables,

321

                       int64_t* numOfNormalTables);

322

  int32_t (*getDBSize)(void* pVnode, SDbSizeStatisInfo* pInfo);

323

324

  SMCtbCursor* (*openCtbCursor)(void* pVnode, tb_uid_t uid, int lock);

325

  int32_t (*resumeCtbCursor)(SMCtbCursor* pCtbCur, int8_t first);

326

  void (*pauseCtbCursor)(SMCtbCursor* pCtbCur);

327

  void (*closeCtbCursor)(SMCtbCursor* pCtbCur);

328

  tb_uid_t (*ctbCursorNext)(SMCtbCursor* pCur);

329

} SStoreMeta;

330

331

typedef struct SStoreMetaReader {

332

  void (*initReader)(SMetaReader* pReader, void* pVnode, int32_t flags, SStoreMeta* pAPI);

333

  void (*clearReader)(SMetaReader* pReader);

334

  void (*readerReleaseLock)(SMetaReader* pReader);

335

  int32_t (*getTableEntryByUid)(SMetaReader* pReader, tb_uid_t uid);

336

  int     (*getTableEntryByVersionUid)(SMetaReader *pReader, int64_t version, tb_uid_t uid);

337

  int32_t (*getTableEntryByName)(SMetaReader* pReader, const char* name);

338

  int32_t (*getEntryGetUidCache)(SMetaReader* pReader, tb_uid_t uid);

339

} SStoreMetaReader;

340

341

typedef struct SUpdateInfo {

342

  SArray*      pTsBuckets;

343

  uint64_t     numBuckets;

344

  SArray*      pTsSBFs;

345

  uint64_t     numSBFs;

346

  int64_t      interval;

347

  int64_t      watermark;

348

  TSKEY        minTS;

349

  SScalableBf* pCloseWinSBF;

350

  SHashObj*    pMap;

351

  int64_t      maxDataVersion;

352

  int8_t       pkColType;

353

  int32_t      pkColLen;

354

  char*        pKeyBuff;

355

  char*        pValueBuff;

356

357

  int (*comparePkRowFn)(void* pValue1, void* pTs, void* pPkVal, __compar_fn_t cmpPkFn);

358

  __compar_fn_t comparePkCol;

359

} SUpdateInfo;

360

361

typedef struct SRecDataInfo {

362

  STimeWindow calWin;

363

  uint64_t    tableUid;

364

  int64_t     dataVersion;

365

  EStreamType mode;

366

  char        pPkColData[];

367

} SRecDataInfo;

368

369

typedef struct SScanRange {

370

  STimeWindow win;

371

  STimeWindow calWin;

372

  SSHashObj*  pGroupIds;

373

  SSHashObj*  pUIds;

374

} SScanRange;

375

376

typedef struct SResultWindowInfo {

377

  SRowBuffPos* pStatePos;

378

  SSessionKey  sessionWin;

379

  bool         isOutput;

380

} SResultWindowInfo;

381

382

typedef struct {

383

  void*   iter;      //  rocksdb_iterator_t*    iter;

384

  void*   snapshot;  //  rocksdb_snapshot_t*    snapshot;

385

  void*   readOpt;   //  rocksdb_readoptions_t* readOpt;

386

  void*   db;        //  rocksdb_t*             db;

387

  void*   pCur;

388

  int64_t number;

389

  void*   pStreamFileState;

390

  int32_t buffIndex;

391

  int32_t hashIter;

392

  void*   pHashData;

393

  int64_t minGpId;

394

} SStreamStateCur;

395

396

typedef struct STableTsDataState {

397

  SSHashObj*       pTableTsDataMap;

398

  __compar_fn_t    comparePkColFn;

399

  void*            pPkValBuff;

400

  int32_t          pkValLen;

401

  SStreamState*    pState;

402

  int32_t          curRecId;

403

  void*            pStreamTaskState;

404

  SArray*          pScanRanges;

405

  SRecDataInfo*    pRecValueBuff;

406

  int32_t          recValueLen;

407

  SStreamStateCur* pRecCur;

408

  int32_t          cfgIndex;

409

  void*            pBatch;

410

  int32_t          batchBufflen;

411

  void*            pBatchBuff;

412

} STableTsDataState;

413

414

typedef struct SStateStore {

415

  int32_t (*streamStatePutParName)(SStreamState* pState, int64_t groupId, const char* tbname);

416

  int32_t (*streamStateGetParName)(SStreamState* pState, int64_t groupId, void** pVal, bool onlyCache,

417

                                   int32_t* pWinCode);

418

  int32_t (*streamStateDeleteParName)(SStreamState* pState, int64_t groupId);

419

  void (*streamStateSetParNameInvalid)(SStreamState* pState);

420

421

  int32_t (*streamStateAddIfNotExist)(SStreamState* pState, const SWinKey* pKey, void** pVal, int32_t* pVLen,

422

                                      int32_t* pWinCode);

423

  void (*streamStateReleaseBuf)(SStreamState* pState, void* pVal, bool used);

424

  void (*streamStateClearBuff)(SStreamState* pState, void* pVal);

425

  void (*streamStateFreeVal)(void* val);

426

  int32_t (*streamStateGetPrev)(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal,

427

                                int32_t* pVLen, int32_t* pWinCode);

428

  int32_t (*streamStateGetAllPrev)(SStreamState* pState, const SWinKey* pKey, SArray* pResArray, int32_t maxNum);

429

430

  int32_t (*streamStatePut)(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);

431

  int32_t (*streamStateGet)(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen, int32_t* pWinCode);

432

  bool (*streamStateCheck)(SStreamState* pState, const SWinKey* key, bool hasLimit, bool* pIsLast);

433

  bool (*streamStateCheckSessionState)(SStreamState* pState, SSessionKey* pKey, TSKEY gap, bool* pIsLast);

434

  int32_t (*streamStateGetByPos)(SStreamState* pState, void* pos, void** pVal);

435

  void (*streamStateDel)(SStreamState* pState, const SWinKey* key);

436

  void (*streamStateDelByGroupId)(SStreamState* pState, uint64_t groupId);

437

  void (*streamStateClear)(SStreamState* pState);

438

  void (*streamStateSetNumber)(SStreamState* pState, int32_t number, int32_t tsIdex);

439

  void (*streamStateSaveInfo)(SStreamState* pState, void* pKey, int32_t keyLen, void* pVal, int32_t vLen);

440

  int32_t (*streamStateGetInfo)(SStreamState* pState, void* pKey, int32_t keyLen, void** pVal, int32_t* pLen);

441

  int32_t (*streamStateGetNumber)(SStreamState* pState);

442

  int32_t (*streamStateDeleteInfo)(SStreamState* pState, void* pKey, int32_t keyLen);

443

444

  int32_t (*streamStateFillPut)(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);

445

  int32_t (*streamStateFillGet)(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen,

446

                                int32_t* pWinCode);

447

  int32_t (*streamStateFillAddIfNotExist)(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen,

448

                                          int32_t* pWinCode);

449

  void (*streamStateFillDel)(SStreamState* pState, const SWinKey* key);

450

  int32_t (*streamStateFillGetNext)(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal,

451

                                    int32_t* pVLen, int32_t* pWinCode);

452

  int32_t (*streamStateFillGetPrev)(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal,

453

                                    int32_t* pVLen, int32_t* pWinCode);

454

455

  void (*streamStateCurNext)(SStreamState* pState, SStreamStateCur* pCur);

456

  void (*streamStateCurPrev)(SStreamState* pState, SStreamStateCur* pCur);

457

458

  SStreamStateCur* (*streamStateGetAndCheckCur)(SStreamState* pState, SWinKey* key);

459

  SStreamStateCur* (*streamStateSeekKeyNext)(SStreamState* pState, const SWinKey* key);

460

  SStreamStateCur* (*streamStateFillSeekKeyNext)(SStreamState* pState, const SWinKey* key);

461

  SStreamStateCur* (*streamStateFillSeekKeyPrev)(SStreamState* pState, const SWinKey* key);

462

  void (*streamStateFreeCur)(SStreamStateCur* pCur);

463

464

  int32_t (*streamStateFillGetGroupKVByCur)(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);

465

  int32_t (*streamStateGetKVByCur)(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);

466

467

  void (*streamStateClearExpiredState)(SStreamState* pState, int32_t numOfKeep, TSKEY minTs);

468

  void (*streamStateClearExpiredSessionState)(SStreamState* pState, int32_t numOfKeep, TSKEY minTs, SSHashObj* pFlushGroup);

469

  int32_t (*streamStateSetRecFlag)(SStreamState* pState, const void* pKey, int32_t keyLen, int32_t mode);

470

  int32_t (*streamStateGetRecFlag)(SStreamState* pState, const void* pKey, int32_t keyLen, int32_t* pMode);

471

472

  int32_t (*streamStateSessionAddIfNotExist)(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal,

473

                                             int32_t* pVLen, int32_t* pWinCode);

474

  int32_t (*streamStateSessionPut)(SStreamState* pState, const SSessionKey* key, void* value, int32_t vLen);

475

  int32_t (*streamStateSessionGet)(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen,

476

                                   int32_t* pWinCode);

477

  void (*streamStateSessionDel)(SStreamState* pState, const SSessionKey* key);

478

  void (*streamStateSessionReset)(SStreamState* pState, void* pVal);

479

  void (*streamStateSessionClear)(SStreamState* pState);

480

  int32_t (*streamStateSessionGetKVByCur)(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen);

481

  int32_t (*streamStateStateAddIfNotExist)(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,

482

                                           state_key_cmpr_fn fn, void** pVal, int32_t* pVLen, int32_t* pWinCode);

483

  int32_t (*streamStateSessionGetKeyByRange)(SStreamState* pState, const SSessionKey* range, SSessionKey* curKey);

484

  int32_t (*streamStateCountGetKeyByRange)(SStreamState* pState, const SSessionKey* range, SSessionKey* curKey);

485

  int32_t (*streamStateSessionAllocWinBuffByNextPosition)(SStreamState* pState, SStreamStateCur* pCur,

486

                                                          const SSessionKey* pKey, void** pVal, int32_t* pVLen);

487

  int32_t (*streamStateSessionSaveToDisk)(STableTsDataState* pTblState, SSessionKey* pKey, SRecDataInfo* pVal, int32_t vLen);

488

  int32_t (*streamStateFlushReaminInfoToDisk)(STableTsDataState* pTblState);

489

  int32_t (*streamStateSessionDeleteAll)(SStreamState* pState);

490

491

  int32_t (*streamStateCountWinAddIfNotExist)(SStreamState* pState, SSessionKey* pKey, COUNT_TYPE winCount,

492

                                              void** ppVal, int32_t* pVLen, int32_t* pWinCode);

493

  int32_t (*streamStateCountWinAdd)(SStreamState* pState, SSessionKey* pKey, COUNT_TYPE winCount, void** pVal,

494

                                    int32_t* pVLen);

495

496

  int32_t (*updateInfoInit)(int64_t interval, int32_t precision, int64_t watermark, bool igUp, int8_t pkType,

497

                            int32_t pkLen, SUpdateInfo** ppInfo);

498

  int32_t (*updateInfoFillBlockData)(SUpdateInfo* pInfo, SSDataBlock* pBlock, int32_t primaryTsCol,

499

                                     int32_t primaryKeyCol, TSKEY* pMaxResTs);

500

  bool (*updateInfoIsUpdated)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts, void* pPkVal, int32_t len);

501

  bool (*updateInfoIsTableInserted)(SUpdateInfo* pInfo, int64_t tbUid);

502

  bool (*isIncrementalTimeStamp)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts, void* pPkVal, int32_t len);

503

504

  void (*updateInfoDestroy)(SUpdateInfo* pInfo);

505

  void (*windowSBfDelete)(SUpdateInfo* pInfo, uint64_t count);

506

  int32_t (*windowSBfAdd)(SUpdateInfo* pInfo, uint64_t count);

507

508

  int32_t (*updateInfoInitP)(SInterval* pInterval, int64_t watermark, bool igUp, int8_t pkType, int32_t pkLen,

509

                             SUpdateInfo** ppInfo);

510

  void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo);

511

  void (*updateInfoDestoryColseWinSBF)(SUpdateInfo* pInfo);

512

  int32_t (*updateInfoSerialize)(SEncoder* pEncoder, const SUpdateInfo* pInfo);

513

  int32_t (*updateInfoDeserialize)(SDecoder* pDeCoder, SUpdateInfo* pInfo);

514

515

  SStreamStateCur* (*streamStateSessionSeekKeyPrev)(SStreamState* pState, const SSessionKey* key);

516

  SStreamStateCur* (*streamStateSessionSeekKeyNext)(SStreamState* pState, const SSessionKey* key);

517

  SStreamStateCur* (*streamStateCountSeekKeyPrev)(SStreamState* pState, const SSessionKey* pKey, COUNT_TYPE count);

518

  SStreamStateCur* (*streamStateSessionSeekKeyCurrentPrev)(SStreamState* pState, const SSessionKey* key);

519

  SStreamStateCur* (*streamStateSessionSeekKeyCurrentNext)(SStreamState* pState, const SSessionKey* key);

520

521

  int32_t (*streamFileStateInit)(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize,

522

                                 GetTsFun fp, void* pFile, TSKEY delMark, const char* id, int64_t ckId, int8_t type,

523

                                 struct SStreamFileState** ppFileState);

524

525

  int32_t (*streamStateGroupPut)(SStreamState* pState, int64_t groupId, void* value, int32_t vLen);

526

  SStreamStateCur* (*streamStateGroupGetCur)(SStreamState* pState);

527

  void (*streamStateGroupCurNext)(SStreamStateCur* pCur);

528

  int32_t (*streamStateGroupGetKVByCur)(SStreamStateCur* pCur, int64_t* pKey, void** pVal, int32_t* pVLen);

529

530

  void (*streamFileStateDestroy)(struct SStreamFileState* pFileState);

531

  void (*streamFileStateClear)(struct SStreamFileState* pFileState);

532

  bool (*needClearDiskBuff)(struct SStreamFileState* pFileState);

533

534

  SStreamState* (*streamStateOpen)(const char* path, void* pTask, int64_t streamId, int32_t taskId);

535

  SStreamState* (*streamStateRecalatedOpen)(const char* path, void* pTask, int64_t streamId, int32_t taskId);

536

  void (*streamStateClose)(SStreamState* pState, bool remove);

537

  int32_t (*streamStateBegin)(SStreamState* pState);

538

  void (*streamStateCommit)(SStreamState* pState);

539

  void (*streamStateDestroy)(SStreamState* pState, bool remove);

540

  void (*streamStateReloadInfo)(SStreamState* pState, TSKEY ts);

541

  void (*streamStateCopyBackend)(SStreamState* src, SStreamState* dst);

542

543

  int32_t (*streamStateGetAndSetTsData)(STableTsDataState* pState, uint64_t tableUid, TSKEY* pCurTs, void** ppCurPkVal,

544

                                        TSKEY lastTs, void* pLastPkVal, int32_t lastPkLen, int32_t* pWinCode);

545

  int32_t (*streamStateTsDataCommit)(STableTsDataState* pState);

546

  int32_t (*streamStateInitTsDataState)(STableTsDataState** ppTsDataState, int8_t pkType, int32_t pkLen, void* pState, void* pOtherState);

547

  void (*streamStateDestroyTsDataState)(STableTsDataState* pTsDataState);

548

  int32_t (*streamStateRecoverTsData)(STableTsDataState* pTsDataState);

549

  int32_t (*streamStateReloadTsDataState)(STableTsDataState* pTsDataState);

550

  int32_t (*streamStateMergeAndSaveScanRange)(STableTsDataState* pTsDataState, STimeWindow* pWin, uint64_t gpId,

551

                                              SRecDataInfo* pRecData, int32_t len);

552

  int32_t (*streamStateMergeAllScanRange)(STableTsDataState* pTsDataState);

553

  int32_t (*streamStatePopScanRange)(STableTsDataState* pTsDataState, SScanRange* pRange);

554

555

  SStreamStateCur* (*streamStateGetLastStateCur)(SStreamState* pState);

556

  void (*streamStateLastStateCurNext)(SStreamStateCur* pCur);

557

  int32_t (*streamStateNLastStateGetKVByCur)(SStreamStateCur* pCur, int32_t num, SArray* pRes);

558

  SStreamStateCur* (*streamStateGetLastSessionStateCur)(SStreamState* pState);

559

  void (*streamStateLastSessionStateCurNext)(SStreamStateCur* pCur);

560

  int32_t (*streamStateNLastSessionStateGetKVByCur)(SStreamStateCur* pCur, int32_t num, SArray* pRes);

561

} SStateStore;

562

563

typedef struct SStorageAPI {

564

  SStoreMeta          metaFn;  // todo: refactor

565

  TsdReader           tsdReader;

566

  SStoreMetaReader    metaReaderFn;

567

  SStoreCacheReader   cacheFn;

568

  SStoreSnapshotFn    snapshotFn;

569

  SStoreTqReader      tqReaderFn;

570

  SStateStore         stateStore;

571

  SMetaDataFilterAPI  metaFilter;

572

  SFunctionStateStore functionStore;

573

} SStorageAPI;

574

575

#ifdef __cplusplus

576

577

#endif

578

579

#endif  // TDENGINE_STORAGEAPI_H

taosdata / TDengine / #4881

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4881

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous