#3809

Committed 01 Apr 2025 03:03AM UTC coverage: 34.048% (+0.02%) from 34.033%

Build # #3809

Build Type

push

travis-ci

Committed by

happyguoxy

Commit Message

test:alter gcda dir

Run Details

148452 of 599532 branches covered (24.76%)

Branch coverage included in aggregate %.

222312 of 489411 relevant lines covered (45.42%)

761122.82 hits per line

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

56.15

/source/libs/stream/test/backendTest.cpp

#include <gtest/gtest.h>

#include <taoserror.h>
#include <tglobal.h>
#include <iostream>
#include <vector>
#include "streamBackendRocksdb.h"
#include "streamSnapshot.h"
#include "streamState.h"
#include "tstream.h"
#include "tstreamFileState.h"
#include "tstreamUpdate.h"

#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
#pragma GCC diagnostic ignored "-Wunused-variable"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wformat"
#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
#pragma GCC diagnostic ignored "-Wpointer-arith"

class BackendEnv : public ::testing::Test {
 protected:
  virtual void SetUp() {}
  virtual void TearDown() {}
};

void *backendCreate() {
  const char *streamPath = "/tmp";
  void       *p = NULL;

  // char *absPath = NULL;
  // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
  // STaskDbWrapper *p = taskDbOpen((char *)streamPath, (char *)"stream-backend", -1);
  // ASSERT(p != NULL);
  return p;
}

SStreamState *stateCreate(const char *path) {
  SStreamTask *pTask = (SStreamTask *)taosMemoryCalloc(1, sizeof(SStreamTask));
  pTask->ver = 1024;
  pTask->id.streamId = 1023;
  pTask->id.taskId = 1111111;
  SStreamMeta *pMeta = NULL;

  int32_t code = streamMetaOpen((path), NULL, NULL, NULL, 0, 0, NULL, &pMeta);
  pTask->pMeta = pMeta;

  SStreamState *p = streamStateOpen((char *)path, pTask, 0, 0);
  ASSERT(p != NULL);
  return p;
}
void *backendOpen() {
  streamMetaInit();
  const char   *path = "/tmp/backend";
  SStreamState *p = stateCreate(path);
  ASSERT(p != NULL);

  // write bacth
  // default/state/fill/sess/func/parname/partag
  int32_t              size = 100;
  std::vector<int64_t> tsArray;

  for (int32_t i = 0; i < size; i++) {
    int64_t ts = taosGetTimestampMs();
    SWinKey key;  // = {.groupId = (uint64_t)(i), .ts = ts};
    key.groupId = (uint64_t)(i);
    key.ts = ts;
    const char *val = "value data";
    int32_t     vlen = strlen(val);
    int32_t     code = streamStatePut_rocksdb(p, &key, (char *)val, vlen);
    ASSERT(code == 0);

    tsArray.push_back(ts);
  }
  for (int32_t i = 0; i < size; i++) {
    int64_t ts = tsArray[i];
    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
    key.groupId = (uint64_t)(i);
    key.ts = ts;

    const char *val = "value data";
    int32_t     len = 0;
    char       *newVal = NULL;
    int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
    ASSERT(code == 0);

    ASSERT(len == strlen(val));
  }
  int64_t ts = tsArray[0];
  SWinKey key = {0};  // {.groupId = (uint64_t)(0), .ts = ts};
  key.groupId = (uint64_t)(0);
  key.ts = ts;

  int32_t code = streamStateDel_rocksdb(p, &key);
  ASSERT(code == 0);

  code = streamStateClear_rocksdb(p);
  ASSERT(code == 0);

  for (int i = 0; i < size; i++) {
    int64_t ts = tsArray[i];
    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
    key.groupId = (uint64_t)(i);
    key.ts = ts;

    const char *val = "value data";
    int32_t     len = 0;
    char       *newVal = NULL;
    int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
    ASSERT(code != 0);
  }
  tsArray.clear();

  for (int i = 0; i < size; i++) {
    int64_t ts = taosGetTimestampMs();
    tsArray.push_back(ts);

    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
    key.groupId = (uint64_t)(i);
    key.ts = ts;

    const char *val = "value data";
    int32_t     vlen = strlen(val);
    code = streamStatePut_rocksdb(p, &key, (char *)val, vlen);
    ASSERT(code == 0);
  }

  SWinKey winkey;
  code = streamStateGetFirst_rocksdb(p, &key);
  ASSERT(code == 0);
  ASSERT(key.ts == tsArray[0]);

  SStreamStateCur *pCurr = streamStateSeekToLast_rocksdb(p);
  ASSERT(pCurr != NULL);
  streamStateFreeCur(pCurr);

  winkey.groupId = 0;
  winkey.ts = tsArray[0];
  char   *val = NULL;
//  int32_t len = 0;

  pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
  ASSERT(pCurr != NULL);

  streamStateFreeCur(pCurr);

  tsArray.clear();
  for (int i = 0; i < size; i++) {
    int64_t ts = taosGetTimestampMs();
    tsArray.push_back(ts);
    STupleKey key = {0};
    key.groupId = (uint64_t)(0);  //= {.groupId = (uint64_t)(0), .ts = ts, .exprIdx = i};
    key.ts = ts;
    key.exprIdx = i;

    const char *val = "Value";
    int32_t     len = strlen(val);
    code = streamStateFuncPut_rocksdb(p, &key, val, len);
    ASSERT(code == 0);
  }
  for (int i = 0; i < size; i++) {
    STupleKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
    key.groupId = (uint64_t)(0);
    key.ts = tsArray[i];
    key.exprIdx = i;

    char   *val = NULL;
    int32_t len = 0;
    int32_t code = streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
    ASSERT(code == 0);

    ASSERT(len == strlen("Value"));
  }
  for (int i = 0; i < size; i++) {
    STupleKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
    key.groupId = (uint64_t)(0);
    key.ts = tsArray[i];
    key.exprIdx = i;

    char   *val = NULL;
    int32_t len = 0;
    int32_t code = streamStateFuncDel_rocksdb(p, &key);
    ASSERT(code == 0);
  }

  // session put
  tsArray.clear();

  for (int i = 0; i < size; i++) {
    SSessionKey key = {0};  //{.win = {.skey = i, .ekey = i}, .groupId = (uint64_t)(0)};
    key.win.skey = i;
    key.win.ekey = i;
    key.groupId = (uint64_t)(0);
    tsArray.push_back(i);

    const char *val = "Value";
    int32_t     len = strlen(val);
    code = streamStateSessionPut_rocksdb(p, &key, val, len);
    ASSERT(code == 0);

    char *pval = NULL;
    ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));
    ASSERT(strncmp(pval, val, len) == 0);
  }

  for (int i = 0; i < size; i++) {
    SSessionKey key = {0};  //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
    key.win.skey = tsArray[i];
    key.win.ekey = tsArray[i];
    key.groupId = (uint64_t)(0);

    const char *val = "Value";
    int32_t     len = strlen(val);

    char *pval = NULL;
    ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));
    ASSERT(strncmp(pval, val, len) == 0);
    taosMemoryFreeClear(pval);
  }

  pCurr = streamStateSessionSeekToLast_rocksdb(p, 0);
  ASSERT(pCurr != NULL);

  {
    SSessionKey key;
    memset(&key, 0, sizeof(key));
    char   *val = NULL;
    int32_t vlen = 0;
    code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);
    ASSERT(code == 0);
    pCurr = streamStateSessionSeekKeyPrev_rocksdb(p, &key);

    code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);
    ASSERT(code == 0);

    ASSERT(key.groupId == 0 && key.win.ekey == tsArray[tsArray.size() - 2]);

    pCurr = streamStateSessionSeekKeyNext_rocksdb(p, &key);
    code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);
    ASSERT(code == 0);
    ASSERT(vlen == strlen("Value"));
    ASSERT(key.groupId == 0 && key.win.skey == tsArray[tsArray.size() - 1]);

    ASSERT(0 == streamStateSessionAddIfNotExist_rocksdb(p, &key, 10, (void **)&val, &vlen));

    ASSERT(0 ==
           streamStateStateAddIfNotExist_rocksdb(p, &key, (char *)"key", strlen("key"), NULL, (void **)&val, &vlen));
  }

  for (int i = 0; i < size; i++) {
    SSessionKey key = {0};  //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
    key.win.skey = tsArray[i];
    key.win.ekey = tsArray[i];
    key.groupId = (uint64_t)(0);

    const char *val = "Value";
    int32_t     len = strlen(val);

    char *pval = NULL;
    ASSERT(0 == streamStateSessionDel_rocksdb(p, &key));
  }

  for (int i = 0; i < size; i++) {
    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
    key.groupId = (uint64_t)(i);
    key.ts = tsArray[i];
    const char *val = "Value";
    int32_t     vlen = strlen(val);
    ASSERT(streamStateFillPut_rocksdb(p, &key, val, vlen) == 0);
  }
  for (int i = 0; i < size; i++) {
    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
    key.groupId = (uint64_t)(i);
    key.ts = tsArray[i];
    char   *val = NULL;
    int32_t vlen = 0;
    ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
    taosMemoryFreeClear(val);
  }
  {
    SWinKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[0]};
    key.groupId = (uint64_t)(0);
    key.ts = tsArray[0];
    SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
    ASSERT(pCurr != NULL);

    char   *val = NULL;
    int32_t vlen = 0;
    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
    ASSERT(vlen == strlen("Value"));
    streamStateFreeCur(pCurr);

    pCurr = streamStateFillSeekKeyNext_rocksdb(p, &key);
    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
    ASSERT(vlen == strlen("Value") && key.groupId == 1 && key.ts == tsArray[1]);

    key.groupId = 1;
    key.ts = tsArray[1];

    pCurr = streamStateFillSeekKeyPrev_rocksdb(p, &key);
    ASSERT(pCurr != NULL);
    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));

    ASSERT(vlen == strlen("Value") && key.groupId == 0 && key.ts == tsArray[0]);
  }

  for (int i = 0; i < size - 1; i++) {
    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
    key.groupId = (uint64_t)(i);
    key.ts = tsArray[i];
    char   *val = NULL;
    int32_t vlen = 0;
    ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
    taosMemoryFreeClear(val);
  }
  streamStateSessionClear_rocksdb(p);

  for (int i = 0; i < size; i++) {
    char tbname[TSDB_TABLE_NAME_LEN] = {0};
    sprintf(tbname, "%s_%d", "tbname", i);
    ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));
  }
  for (int i = 0; i < size; i++) {
    char *val = NULL;
    ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));
    ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);
    taosMemoryFree(val);
  }

  for (int i = 0; i < size; i++) {
    char tbname[TSDB_TABLE_NAME_LEN] = {0};
    sprintf(tbname, "%s_%d", "tbname", i);
    ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));
  }
  for (int i = 0; i < size; i++) {
    char *val = NULL;
    ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));
    ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);
    taosMemoryFree(val);
  }
  for (int i = 0; i < size; i++) {
    char key[128] = {0};
    sprintf(key, "tbname_%d", i);
    char val[128] = {0};
    sprintf(val, "val_%d", i);
    code = streamDefaultPut_rocksdb(p, key, val, strlen(val));
    ASSERT(code == 0);
  }
  for (int i = 0; i < size; i++) {
    char key[128] = {0};
    sprintf(key, "tbname_%d", i);

    char   *val = NULL;
    int32_t len = 0;
    code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
    ASSERT(code == 0);
  }
  SArray *result = taosArrayInit(8, sizeof(void *));
  code = streamDefaultIterGet_rocksdb(p, "tbname", "tbname_99", result);
  ASSERT(code == 0);

  ASSERT(taosArrayGetSize(result) >= 0);

  return p;
  // streamStateClose((SStreamState *)p, true);
}
TEST_F(BackendEnv, checkOpen) {
  SStreamState *p = (SStreamState *)backendOpen();
  int64_t       tsStart = taosGetTimestampMs();
  {
    void   *pBatch = streamStateCreateBatch();
    int32_t size = 0;
    for (int i = 0; i < size; i++) {
      char key[128] = {0};
      sprintf(key, "key_%d", i);
      char val[128] = {0};
      sprintf(val, "val_%d", i);
      int32_t code = streamStatePutBatch(p, "default", (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
                                         (int32_t)(strlen(val)), tsStart + 100000);
      ASSERT(code == 0);
    }

    int32_t code = streamStatePutBatch_rocksdb(p, pBatch);
    ASSERT(code == 0);

    streamStateDestroyBatch(pBatch);
  }
  {
    void   *pBatch = streamStateCreateBatch();
    int32_t size = 0;
    char    valBuf[256] = {0};
    for (int i = 0; i < size; i++) {
      char key[128] = {0};
      sprintf(key, "key_%d", i);
      char val[128] = {0};
      sprintf(val, "val_%d", i);
      int32_t code = streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
                                                 (int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);
      ASSERT(code == 0);
    }
    int32_t code = streamStatePutBatch_rocksdb(p, pBatch);
    ASSERT(code == 0);
    streamStateDestroyBatch(pBatch);
  }

  SArray* pList = taosArrayInit(3, sizeof(int64_t));

  // do checkpoint 2
  int32_t code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2, 0, pList);
  ASSERT(code == 0);

  {
    void   *pBatch = streamStateCreateBatch();
    int32_t size = 0;
    char    valBuf[256] = {0};
    for (int i = 0; i < size; i++) {
      char key[128] = {0};
      sprintf(key, "key_%d", i);
      char val[128] = {0};
      sprintf(val, "val_%d", i);
      int32_t code = streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
                                                 (int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);
      ASSERT(code == 0);
    }
    code = streamStatePutBatch_rocksdb(p, pBatch);
    ASSERT(code == 0);

    streamStateDestroyBatch(pBatch);
  }

  taosArrayClear(pList);

  code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 3, 0, pList);
  ASSERT(code == 0);

  const char *path = "/tmp/backend/stream";
  const char *dump = "/tmp/backend/stream/dump";
  // taosMkDir(dump);
  taosMulMkDir(dump);
  SBkdMgt *mgt = NULL;

  code = bkdMgtCreate((char *)path, &mgt);
  SArray *result = taosArrayInit(4, sizeof(void *));
  bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);

  taosArrayClear(pList);
  code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 4, 0, pList);
  ASSERT(code == 0);

  taosArrayDestroy(pList);
  taosArrayClear(result);
  code = bkdMgtGetDelta(mgt, p->pTdbState->idstr, 4, result, (char *)dump);
  ASSERT(code == 0);

  bkdMgtDestroy(mgt);
  streamStateClose((SStreamState *)p, true);
  // {
  //   taosRemoveDir("/tmp/backend");
  //   const char *  path = "/tmp/backend";
  //   SStreamState *p = stateCreate(path);
  // }
  taosRemoveDir(path);
  // streamStateClose((SStreamState *)p, true);
}

TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }

typedef struct BdKV {
  uint32_t k;
  uint32_t v;
} BdKV;

BdKV kvDict[] = {{0, 2},     {1, 2},     {15, 16},     {31, 32},     {56, 64},    {100, 128},
                 {200, 256}, {500, 512}, {1000, 1024}, {2000, 2048}, {3000, 4096}};

TEST_F(BackendEnv, backendUtil) {
  for (int i = 0; i < sizeof(kvDict) / sizeof(kvDict[0]); i++) {
    ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v);
  }
}
TEST_F(BackendEnv, oldBackendInit) {
  const char *path = "/tmp/backend1";
  int32_t     code = taosMulMkDir(path);
  ASSERT(code == 0);

  {
    SBackendWrapper *p = NULL;
    int32_t code = streamBackendInit(path, 10, 10, &p);
    streamBackendCleanup((void *)p);
  }
  {
    SBackendWrapper *p = NULL;
    int32_t code = streamBackendInit(path, 10, 10, &p);
    streamBackendCleanup((void *)p);
  }

  taosRemoveDir(path);
}

int main(int argc, char **argv) {
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}

1	#include <gtest/gtest.h>
2
3	#include <taoserror.h>
4	#include <tglobal.h>
5	#include <iostream>
6	#include <vector>
7	#include "streamBackendRocksdb.h"
8	#include "streamSnapshot.h"
9	#include "streamState.h"
10	#include "tstream.h"
11	#include "tstreamFileState.h"
12	#include "tstreamUpdate.h"
13
14	#pragma GCC diagnostic push
15	#pragma GCC diagnostic ignored "-Wwrite-strings"
16	#pragma GCC diagnostic ignored "-Wunused-function"
17	#pragma GCC diagnostic ignored "-Wunused-variable"
18	#pragma GCC diagnostic ignored "-Wsign-compare"
19	#pragma GCC diagnostic ignored "-Wsign-compare"
20	#pragma GCC diagnostic ignored "-Wformat"
21	#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
22	#pragma GCC diagnostic ignored "-Wpointer-arith"
23
24	class BackendEnv : public ::testing::Test {
25	protected:
26	virtual void SetUp() {}	4✔
27	virtual void TearDown() {}	4✔
28	};
29
30	void *backendCreate() {	×
31	const char *streamPath = "/tmp";	×
32	void *p = NULL;	×
33
34	// char *absPath = NULL;
35	// // SBackendWrapper p = (SBackendWrapper )streamBackendInit(streamPath, -1, 2);
36	// STaskDbWrapper p = taskDbOpen((char )streamPath, (char *)"stream-backend", -1);
37	// ASSERT(p != NULL);
38	return p;	×
39	}
40
41	SStreamState stateCreate(const char path) {	1✔
42	SStreamTask pTask = (SStreamTask )taosMemoryCalloc(1, sizeof(SStreamTask));	1!
43	pTask->ver = 1024;	1✔
44	pTask->id.streamId = 1023;	1✔
45	pTask->id.taskId = 1111111;	1✔
46	SStreamMeta *pMeta = NULL;	1✔
47
48	int32_t code = streamMetaOpen((path), NULL, NULL, NULL, 0, 0, NULL, &pMeta);	1!
49	pTask->pMeta = pMeta;	1✔
50
51	SStreamState p = streamStateOpen((char )path, pTask, 0, 0);	1!
52	ASSERT(p != NULL);	1!
53	return p;	1✔
54	}
55	void *backendOpen() {	1✔
56	streamMetaInit();	1!
57	const char *path = "/tmp/backend";	1✔
58	SStreamState *p = stateCreate(path);	1!
59	ASSERT(p != NULL);	1!
60
61	// write bacth
62	// default/state/fill/sess/func/parname/partag
63	int32_t size = 100;	1✔
64	std::vector<int64_t> tsArray;	1✔
65
66	for (int32_t i = 0; i < size; i++) {	101✔
67	int64_t ts = taosGetTimestampMs();	100✔
68	SWinKey key; // = {.groupId = (uint64_t)(i), .ts = ts};
69	key.groupId = (uint64_t)(i);	100✔
70	key.ts = ts;	100✔
71	const char *val = "value data";	100✔
72	int32_t vlen = strlen(val);	100✔
73	int32_t code = streamStatePut_rocksdb(p, &key, (char *)val, vlen);	100!
74	ASSERT(code == 0);	100!
75
76	tsArray.push_back(ts);	100!
77	}
78	for (int32_t i = 0; i < size; i++) {	101✔
79	int64_t ts = tsArray[i];	100✔
80	SWinKey key = {0}; //{.groupId = (uint64_t)(i), .ts = ts};	100✔
81	key.groupId = (uint64_t)(i);	100✔
82	key.ts = ts;	100✔
83
84	const char *val = "value data";	100✔
85	int32_t len = 0;	100✔
86	char *newVal = NULL;	100✔
87	int32_t code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);	100!
88	ASSERT(code == 0);	100!
89
90	ASSERT(len == strlen(val));	100!
91	}
92	int64_t ts = tsArray[0];	1✔
93	SWinKey key = {0}; // {.groupId = (uint64_t)(0), .ts = ts};	1✔
94	key.groupId = (uint64_t)(0);	1✔
95	key.ts = ts;	1✔
96
97	int32_t code = streamStateDel_rocksdb(p, &key);	1!
98	ASSERT(code == 0);	1!
99
100	code = streamStateClear_rocksdb(p);	1!
101	ASSERT(code == 0);	1!
102
103	for (int i = 0; i < size; i++) {	101✔
104	int64_t ts = tsArray[i];	100✔
105	SWinKey key = {0}; //{.groupId = (uint64_t)(i), .ts = ts};	100✔
106	key.groupId = (uint64_t)(i);	100✔
107	key.ts = ts;	100✔
108
109	const char *val = "value data";	100✔
110	int32_t len = 0;	100✔
111	char *newVal = NULL;	100✔
112	int32_t code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);	100!
113	ASSERT(code != 0);	100!
114	}
115	tsArray.clear();	1✔
116
117	for (int i = 0; i < size; i++) {	101✔
118	int64_t ts = taosGetTimestampMs();	100✔
119	tsArray.push_back(ts);	100!
120
121	SWinKey key = {0}; //{.groupId = (uint64_t)(i), .ts = ts};	100✔
122	key.groupId = (uint64_t)(i);	100✔
123	key.ts = ts;	100✔
124
125	const char *val = "value data";	100✔
126	int32_t vlen = strlen(val);	100✔
127	code = streamStatePut_rocksdb(p, &key, (char *)val, vlen);	100!
128	ASSERT(code == 0);	100!
129	}
130
131	SWinKey winkey;
132	code = streamStateGetFirst_rocksdb(p, &key);	1!
133	ASSERT(code == 0);	1!
134	ASSERT(key.ts == tsArray[0]);	1!
135
136	SStreamStateCur *pCurr = streamStateSeekToLast_rocksdb(p);	1!
137	ASSERT(pCurr != NULL);	1!
138	streamStateFreeCur(pCurr);	1!
139
140	winkey.groupId = 0;	1✔
141	winkey.ts = tsArray[0];	1✔
142	char *val = NULL;	1✔
143	// int32_t len = 0;
144
145	pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);	1!
146	ASSERT(pCurr != NULL);	1!
147
148	streamStateFreeCur(pCurr);	1!
149
150	tsArray.clear();	1✔
151	for (int i = 0; i < size; i++) {	101✔
152	int64_t ts = taosGetTimestampMs();	100✔
153	tsArray.push_back(ts);	100!
154	STupleKey key = {0};	100✔
155	key.groupId = (uint64_t)(0); //= {.groupId = (uint64_t)(0), .ts = ts, .exprIdx = i};	100✔
156	key.ts = ts;	100✔
157	key.exprIdx = i;	100✔
158
159	const char *val = "Value";	100✔
160	int32_t len = strlen(val);	100✔
161	code = streamStateFuncPut_rocksdb(p, &key, val, len);	100!
162	ASSERT(code == 0);	100!
163	}
164	for (int i = 0; i < size; i++) {	101✔
165	STupleKey key = {0}; //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};	100✔
166	key.groupId = (uint64_t)(0);	100✔
167	key.ts = tsArray[i];	100✔
168	key.exprIdx = i;	100✔
169
170	char *val = NULL;	100✔
171	int32_t len = 0;	100✔
172	int32_t code = streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);	100!
173	ASSERT(code == 0);	100!
174
175	ASSERT(len == strlen("Value"));	100!
176	}
177	for (int i = 0; i < size; i++) {	101✔
178	STupleKey key = {0}; //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};	100✔
179	key.groupId = (uint64_t)(0);	100✔
180	key.ts = tsArray[i];	100✔
181	key.exprIdx = i;	100✔
182
183	char *val = NULL;	100✔
184	int32_t len = 0;	100✔
185	int32_t code = streamStateFuncDel_rocksdb(p, &key);	100!
186	ASSERT(code == 0);	100!
187	}
188
189	// session put
190	tsArray.clear();	1✔
191
192	for (int i = 0; i < size; i++) {	101✔
193	SSessionKey key = {0}; //{.win = {.skey = i, .ekey = i}, .groupId = (uint64_t)(0)};	100✔
194	key.win.skey = i;	100✔
195	key.win.ekey = i;	100✔
196	key.groupId = (uint64_t)(0);	100✔
197	tsArray.push_back(i);	100!
198
199	const char *val = "Value";	100✔
200	int32_t len = strlen(val);	100✔
201	code = streamStateSessionPut_rocksdb(p, &key, val, len);	100!
202	ASSERT(code == 0);	100!
203
204	char *pval = NULL;	100✔
205	ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));	100!
206	ASSERT(strncmp(pval, val, len) == 0);	100!
207	}
208
209	for (int i = 0; i < size; i++) {	101✔
210	SSessionKey key = {0}; //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};	100✔
211	key.win.skey = tsArray[i];	100✔
212	key.win.ekey = tsArray[i];	100✔
213	key.groupId = (uint64_t)(0);	100✔
214
215	const char *val = "Value";	100✔
216	int32_t len = strlen(val);	100✔
217
218	char *pval = NULL;	100✔
219	ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));	100!
220	ASSERT(strncmp(pval, val, len) == 0);	100!
221	taosMemoryFreeClear(pval);	100!
222	}
223
224	pCurr = streamStateSessionSeekToLast_rocksdb(p, 0);	1!
225	ASSERT(pCurr != NULL);	1!
226
227	{
228	SSessionKey key;
229	memset(&key, 0, sizeof(key));	1✔
230	char *val = NULL;	1✔
231	int32_t vlen = 0;	1✔
232	code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);	1!
233	ASSERT(code == 0);	1!
234	pCurr = streamStateSessionSeekKeyPrev_rocksdb(p, &key);	1!
235
236	code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);	1!
237	ASSERT(code == 0);	1!
238
239	ASSERT(key.groupId == 0 && key.win.ekey == tsArray[tsArray.size() - 2]);	1!
240
241	pCurr = streamStateSessionSeekKeyNext_rocksdb(p, &key);	1!
242	code = streamStateSessionGetKVByCur_rocksdb(NULL, pCurr, &key, (void **)&val, &vlen);	1!
243	ASSERT(code == 0);	1!
244	ASSERT(vlen == strlen("Value"));	1!
245	ASSERT(key.groupId == 0 && key.win.skey == tsArray[tsArray.size() - 1]);	1!
246
247	ASSERT(0 == streamStateSessionAddIfNotExist_rocksdb(p, &key, 10, (void **)&val, &vlen));	1!
248
249	ASSERT(0 ==	1!
250	streamStateStateAddIfNotExist_rocksdb(p, &key, (char )"key", strlen("key"), NULL, (void *)&val, &vlen));
251	}
252
253	for (int i = 0; i < size; i++) {	101✔
254	SSessionKey key = {0}; //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};	100✔
255	key.win.skey = tsArray[i];	100✔
256	key.win.ekey = tsArray[i];	100✔
257	key.groupId = (uint64_t)(0);	100✔
258
259	const char *val = "Value";	100✔
260	int32_t len = strlen(val);	100✔
261
262	char *pval = NULL;	100✔
263	ASSERT(0 == streamStateSessionDel_rocksdb(p, &key));	100!
264	}
265
266	for (int i = 0; i < size; i++) {	101✔
267	SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]};	100✔
268	key.groupId = (uint64_t)(i);	100✔
269	key.ts = tsArray[i];	100✔
270	const char *val = "Value";	100✔
271	int32_t vlen = strlen(val);	100✔
272	ASSERT(streamStateFillPut_rocksdb(p, &key, val, vlen) == 0);	100!
273	}
274	for (int i = 0; i < size; i++) {	101✔
275	SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]};	100✔
276	key.groupId = (uint64_t)(i);	100✔
277	key.ts = tsArray[i];	100✔
278	char *val = NULL;	100✔
279	int32_t vlen = 0;	100✔
280	ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);	100!
281	taosMemoryFreeClear(val);	100!
282	}
283	{
284	SWinKey key = {0}; //{.groupId = (uint64_t)(0), .ts = tsArray[0]};	1✔
285	key.groupId = (uint64_t)(0);	1✔
286	key.ts = tsArray[0];	1✔
287	SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);	1!
288	ASSERT(pCurr != NULL);	1!
289
290	char *val = NULL;	1✔
291	int32_t vlen = 0;	1✔
292	ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));	1!
293	ASSERT(vlen == strlen("Value"));	1!
294	streamStateFreeCur(pCurr);	1!
295
296	pCurr = streamStateFillSeekKeyNext_rocksdb(p, &key);	1!
297	ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));	1!
298	ASSERT(vlen == strlen("Value") && key.groupId == 1 && key.ts == tsArray[1]);	1!
299
300	key.groupId = 1;	1✔
301	key.ts = tsArray[1];	1✔
302
303	pCurr = streamStateFillSeekKeyPrev_rocksdb(p, &key);	1!
304	ASSERT(pCurr != NULL);	1!
305	ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));	1!
306
307	ASSERT(vlen == strlen("Value") && key.groupId == 0 && key.ts == tsArray[0]);	1!
308	}
309
310	for (int i = 0; i < size - 1; i++) {	100✔
311	SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]};	99✔
312	key.groupId = (uint64_t)(i);	99✔
313	key.ts = tsArray[i];	99✔
314	char *val = NULL;	99✔
315	int32_t vlen = 0;	99✔
316	ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);	99!
317	taosMemoryFreeClear(val);	99!
318	}
319	streamStateSessionClear_rocksdb(p);	1!
320
321	for (int i = 0; i < size; i++) {	101✔
322	char tbname[TSDB_TABLE_NAME_LEN] = {0};	100✔
323	sprintf(tbname, "%s_%d", "tbname", i);	100✔
324	ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));	100!
325	}
326	for (int i = 0; i < size; i++) {	101✔
327	char *val = NULL;	100✔
328	ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));	100!
329	ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);	100!
330	taosMemoryFree(val);	100!
331	}
332
333	for (int i = 0; i < size; i++) {	101✔
334	char tbname[TSDB_TABLE_NAME_LEN] = {0};	100✔
335	sprintf(tbname, "%s_%d", "tbname", i);	100✔
336	ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));	100!
337	}
338	for (int i = 0; i < size; i++) {	101✔
339	char *val = NULL;	100✔
340	ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));	100!
341	ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);	100!
342	taosMemoryFree(val);	100!
343	}
344	for (int i = 0; i < size; i++) {	101✔
345	char key[128] = {0};	100✔
346	sprintf(key, "tbname_%d", i);	100✔
347	char val[128] = {0};	100✔
348	sprintf(val, "val_%d", i);	100✔
349	code = streamDefaultPut_rocksdb(p, key, val, strlen(val));	100!
350	ASSERT(code == 0);	100!
351	}
352	for (int i = 0; i < size; i++) {	101✔
353	char key[128] = {0};	100✔
354	sprintf(key, "tbname_%d", i);	100✔
355
356	char *val = NULL;	100✔
357	int32_t len = 0;	100✔
358	code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);	100!
359	ASSERT(code == 0);	100!
360	}
361	SArray result = taosArrayInit(8, sizeof(void ));	1!
362	code = streamDefaultIterGet_rocksdb(p, "tbname", "tbname_99", result);	1!
363	ASSERT(code == 0);	1!
364
365	ASSERT(taosArrayGetSize(result) >= 0);	1!
366
367	return p;	2✔
368	// streamStateClose((SStreamState *)p, true);
369	}
370	TEST_F(BackendEnv, checkOpen) {	4✔
371	SStreamState p = (SStreamState )backendOpen();	1!
372	int64_t tsStart = taosGetTimestampMs();	1✔
373	{
374	void *pBatch = streamStateCreateBatch();	1!
375	int32_t size = 0;	1✔
376	for (int i = 0; i < size; i++) {	1!
377	char key[128] = {0};	×
378	sprintf(key, "key_%d", i);	×
379	char val[128] = {0};	×
380	sprintf(val, "val_%d", i);	×
381	int32_t code = streamStatePutBatch(p, "default", (rocksdb_writebatch_t )pBatch, (void )key, (void *)val,	×
382	(int32_t)(strlen(val)), tsStart + 100000);	×
383	ASSERT(code == 0);	×
384	}
385
386	int32_t code = streamStatePutBatch_rocksdb(p, pBatch);	1!
387	ASSERT(code == 0);	1!
388
389	streamStateDestroyBatch(pBatch);	1!
390	}
391	{
392	void *pBatch = streamStateCreateBatch();	1!
393	int32_t size = 0;	1✔
394	char valBuf[256] = {0};	1✔
395	for (int i = 0; i < size; i++) {	1!
396	char key[128] = {0};	×
397	sprintf(key, "key_%d", i);	×
398	char val[128] = {0};	×
399	sprintf(val, "val_%d", i);	×
400	int32_t code = streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t )pBatch, (void )key, (void *)val,	×
401	(int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);	×
402	ASSERT(code == 0);	×
403	}
404	int32_t code = streamStatePutBatch_rocksdb(p, pBatch);	1!
405	ASSERT(code == 0);	1!
406	streamStateDestroyBatch(pBatch);	1!
407	}
408
409	SArray* pList = taosArrayInit(3, sizeof(int64_t));	1!
410
411	// do checkpoint 2
412	int32_t code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2, 0, pList);	1!
413	ASSERT(code == 0);	1!
414
415	{
416	void *pBatch = streamStateCreateBatch();	1!
417	int32_t size = 0;	1✔
418	char valBuf[256] = {0};	1✔
419	for (int i = 0; i < size; i++) {	1!
420	char key[128] = {0};	×
421	sprintf(key, "key_%d", i);	×
422	char val[128] = {0};	×
423	sprintf(val, "val_%d", i);	×
424	int32_t code = streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t )pBatch, (void )key, (void *)val,	×
425	(int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);	×
426	ASSERT(code == 0);	×
427	}
428	code = streamStatePutBatch_rocksdb(p, pBatch);	1!
429	ASSERT(code == 0);	1!
430
431	streamStateDestroyBatch(pBatch);	1!
432	}
433
434	taosArrayClear(pList);	1!
435
436	code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 3, 0, pList);	1!
437	ASSERT(code == 0);	1!
438
439	const char *path = "/tmp/backend/stream";	1✔
440	const char *dump = "/tmp/backend/stream/dump";	1✔
441	// taosMkDir(dump);
442	taosMulMkDir(dump);	1!
443	SBkdMgt *mgt = NULL;	1✔
444
445	code = bkdMgtCreate((char *)path, &mgt);	1!
446	SArray result = taosArrayInit(4, sizeof(void ));	1!
447	bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);	1!
448
449	taosArrayClear(pList);	1!
450	code = taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 4, 0, pList);	1!
451	ASSERT(code == 0);	1!
452
453	taosArrayDestroy(pList);	1!
454	taosArrayClear(result);	1!
455	code = bkdMgtGetDelta(mgt, p->pTdbState->idstr, 4, result, (char *)dump);	1!
456	ASSERT(code == 0);	1!
457
458	bkdMgtDestroy(mgt);	1!
459	streamStateClose((SStreamState *)p, true);	1!
460	// {
461	// taosRemoveDir("/tmp/backend");
462	// const char * path = "/tmp/backend";
463	// SStreamState *p = stateCreate(path);
464	// }
465	taosRemoveDir(path);	1!
466	// streamStateClose((SStreamState *)p, true);
467	}	1✔
468
469	TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }	4✔
470
471	typedef struct BdKV {
472	uint32_t k;
473	uint32_t v;
474	} BdKV;
475
476	BdKV kvDict[] = {{0, 2}, {1, 2}, {15, 16}, {31, 32}, {56, 64}, {100, 128},
477	{200, 256}, {500, 512}, {1000, 1024}, {2000, 2048}, {3000, 4096}};
478
479	TEST_F(BackendEnv, backendUtil) {	4✔
480	for (int i = 0; i < sizeof(kvDict) / sizeof(kvDict[0]); i++) {	12✔
481	ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v);	11!
482	}
483	}
484	TEST_F(BackendEnv, oldBackendInit) {	4✔
485	const char *path = "/tmp/backend1";	1✔
486	int32_t code = taosMulMkDir(path);	1✔
487	ASSERT(code == 0);	1!
488
489	{
490	SBackendWrapper *p = NULL;	1✔
491	int32_t code = streamBackendInit(path, 10, 10, &p);	1!
492	streamBackendCleanup((void *)p);	1!
493	}
494	{
495	SBackendWrapper *p = NULL;	1✔
496	int32_t code = streamBackendInit(path, 10, 10, &p);	1!
497	streamBackendCleanup((void *)p);	1!
498	}
499
500	taosRemoveDir(path);	1✔
501	}	1✔
502
503	int main(int argc, char **argv) {	1✔
504	testing::InitGoogleTest(&argc, argv);	1✔
505	return RUN_ALL_TESTS();	1✔
506	}

taosdata / TDengine / #3809

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