Merge pull request #1826 from taosdata/feature/2.0tsdb

Feature/2.0tsdb

src/common/inc/tdataformat.h

@@ -97,7 +97,7 @@ typedef void *SDataRow;
 #define TD_DATA_ROW_HEAD_SIZE sizeof(int32_t)
 
 #define dataRowLen(r) (*(int32_t *)(r))
-#define dataRowTuple(r) POINTER_DRIFT(r, TD_DATA_ROW_HEAD_SIZE)
+#define dataRowTuple(r) POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
 #define dataRowKey(r) (*(TSKEY *)(dataRowTuple(r)))
 #define dataRowSetLen(r, l) (dataRowLen(r) = (l))
 #define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
@@ -114,10 +114,10 @@ static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t o
   switch (type) {
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
-      return POINTER_DRIFT(row, *(VarDataOffsetT *)POINTER_DRIFT(row, offset));
+      return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
       break;
     default:
-      return POINTER_DRIFT(row, offset);
+      return POINTER_SHIFT(row, offset);
       break;
   }
 }
@@ -149,11 +149,11 @@ static FORCE_INLINE void *tdGetColDataOfRow(SDataCol *pCol, int row) {
   switch (pCol->type) {
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
-      return POINTER_DRIFT(pCol->pData, pCol->dataOff[row]);
+      return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
       break;
 
     default:
-      return POINTER_DRIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
+      return POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
       break;
   }
 }

src/common/src/tdataformat.c

@@ -167,17 +167,17 @@ void tdFreeDataRow(SDataRow row) {
 int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_t offset) {
   ASSERT(value != NULL);
   int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
-  char *  ptr = POINTER_DRIFT(row, dataRowLen(row));
+  char *  ptr = POINTER_SHIFT(row, dataRowLen(row));
 
   switch (type) {
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
-      *(VarDataOffsetT *)POINTER_DRIFT(row, toffset) = dataRowLen(row);
+      *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
       memcpy(ptr, value, varDataTLen(value));
       dataRowLen(row) += varDataTLen(value);
       break;
     default:
-      memcpy(POINTER_DRIFT(row, toffset), value, TYPE_BYTES[type]);
+      memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
       break;
   }
 
@@ -202,13 +202,13 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
   if (pDataCol->type == TSDB_DATA_TYPE_BINARY || pDataCol->type == TSDB_DATA_TYPE_NCHAR) {
     pDataCol->spaceSize = (sizeof(VarDataLenT) + pDataCol->bytes) * maxPoints;
     pDataCol->dataOff = (VarDataOffsetT *)(*pBuf);
-    pDataCol->pData = POINTER_DRIFT(*pBuf, TYPE_BYTES[pDataCol->type] * maxPoints);
-    *pBuf = POINTER_DRIFT(*pBuf, pDataCol->spaceSize + TYPE_BYTES[pDataCol->type] * maxPoints);
+    pDataCol->pData = POINTER_SHIFT(*pBuf, TYPE_BYTES[pDataCol->type] * maxPoints);
+    *pBuf = POINTER_SHIFT(*pBuf, pDataCol->spaceSize + TYPE_BYTES[pDataCol->type] * maxPoints);
   } else {
     pDataCol->spaceSize = pDataCol->bytes * maxPoints;
     pDataCol->dataOff = NULL;
     pDataCol->pData = *pBuf;
-    *pBuf = POINTER_DRIFT(*pBuf, pDataCol->spaceSize);
+    *pBuf = POINTER_SHIFT(*pBuf, pDataCol->spaceSize);
   }
 
 }
@@ -222,13 +222,13 @@ void dataColAppendVal(SDataCol *pCol, void *value, int numOfPoints, int maxPoint
       // set offset
       pCol->dataOff[numOfPoints] = pCol->len;
       // Copy data
-      memcpy(POINTER_DRIFT(pCol->pData, pCol->len), value, varDataTLen(value));
+      memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, varDataTLen(value));
       // Update the length
       pCol->len += varDataTLen(value);
       break;
     default:
       ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfPoints);
-      memcpy(POINTER_DRIFT(pCol->pData, pCol->len), value, pCol->bytes);
+      memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
       pCol->len += pCol->bytes;
       break;
   }
@@ -244,12 +244,12 @@ void dataColPopPoints(SDataCol *pCol, int pointsToPop, int numOfPoints) {
     VarDataOffsetT toffset = pCol->dataOff[pointsToPop];
     pCol->len = pCol->len - toffset;
     ASSERT(pCol->len > 0);
-    memmove(pCol->pData, POINTER_DRIFT(pCol->pData, toffset), pCol->len);
+    memmove(pCol->pData, POINTER_SHIFT(pCol->pData, toffset), pCol->len);
     dataColSetOffset(pCol, pointsLeft);
   } else {
     ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfPoints);
     pCol->len = TYPE_BYTES[pCol->type] * pointsLeft;
-    memmove(pCol->pData, POINTER_DRIFT(pCol->pData, TYPE_BYTES[pCol->type] * pointsToPop), pCol->len);
+    memmove(pCol->pData, POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * pointsToPop), pCol->len);
   }
 }
 
@@ -301,7 +301,7 @@ void dataColSetOffset(SDataCol *pCol, int nEle) {
   for (int i = 0; i < nEle; i++) {
     pCol->dataOff[i] = offset;
     offset += varDataTLen(tptr);
-    tptr = POINTER_DRIFT(tptr, varDataTLen(tptr));
+    tptr = POINTER_SHIFT(tptr, varDataTLen(tptr));
   }
 }
 

src/tsdb/inc/tsdbMain.h

@@ -21,6 +21,7 @@
 #include "tskiplist.h"
 #include "tutil.h"
 #include "tlog.h"
+#include "tcoding.h"
 
 #ifdef __cplusplus
 extern "C" {

src/tsdb/src/tsdbFile.c

@@ -178,7 +178,7 @@ void tsdbFitRetention(STsdbRepo *pRepo) {
   SFileGroup *pGroup = pFileH->fGroup;
 
   int mfid =
-      tsdbGetKeyFileId(taosGetTimestamp(pRepo->config.precision), pRepo->config.daysPerFile, pRepo->config.precision);
+      tsdbGetKeyFileId(taosGetTimestamp(pRepo->config.precision), pRepo->config.daysPerFile, pRepo->config.precision) - pFileH->maxFGroups + 3;
 
   while (pFileH->numOfFGroups > 0 && pGroup[0].fileId < mfid) {
     tsdbRemoveFileGroup(pFileH, pGroup[0].fileId);

src/tsdb/src/tsdbMeta.c

@@ -535,5 +535,5 @@ static int tsdbEstimateTableEncodeSize(STable *pTable) {
 char *getTupleKey(const void * data) {
   SDataRow row = (SDataRow)data;
 
-  return POINTER_DRIFT(row, TD_DATA_ROW_HEAD_SIZE);
+  return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE);
 }
\ No newline at end of file

src/util/inc/tcoding.h

@@ -24,98 +24,188 @@ extern "C" {
 
 #include "tutil.h"
 
-const int TNUMBER = 1;
-#define IS_LITTLE_ENDIAN() (*(char *)(&TNUMBER) != 0)
+// TODO: move this to a platform file
+#define ENCODE_LIMIT (((uint8_t)1) << 7)
+static const int32_t TNUMBER = 1;
+#define IS_LITTLE_ENDIAN() (*(uint8_t *)(&TNUMBER) != 0)
 
 static FORCE_INLINE void *taosEncodeFixed16(void *buf, uint16_t value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(buf, &value, sizeof(value));
   } else {
-    ((char *)buf)[0] = value & 0xff;
-    ((char *)buf)[1] = (value >> 8) & 0xff;
+    ((uint8_t *)buf)[0] = value & 0xff;
+    ((uint8_t *)buf)[1] = (value >> 8) & 0xff;
   }
 
-  return POINTER_DRIFT(buf, sizeof(value));
+  return POINTER_SHIFT(buf, sizeof(value));
 }
 
 static FORCE_INLINE void *taosEncodeFixed32(void *buf, uint32_t value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(buf, &value, sizeof(value));
   } else {
-    ((char *)buf)[0] = value & 0xff;
-    ((char *)buf)[1] = (value >> 8) & 0xff;
-    ((char *)buf)[2] = (value >> 16) & 0xff;
-    ((char *)buf)[3] = (value >> 24) & 0xff;
+    ((uint8_t *)buf)[0] = value & 0xff;
+    ((uint8_t *)buf)[1] = (value >> 8) & 0xff;
+    ((uint8_t *)buf)[2] = (value >> 16) & 0xff;
+    ((uint8_t *)buf)[3] = (value >> 24) & 0xff;
   }
 
-  return POINTER_DRIFT(buf, sizeof(value));
+  return POINTER_SHIFT(buf, sizeof(value));
 }
 
 static FORCE_INLINE void *taosEncodeFixed64(void *buf, uint64_t value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(buf, &value, sizeof(value));
   } else {
-    ((char *)buf)[0] = value & 0xff;
-    ((char *)buf)[1] = (value >> 8) & 0xff;
-    ((char *)buf)[2] = (value >> 16) & 0xff;
-    ((char *)buf)[3] = (value >> 24) & 0xff;
-    ((char *)buf)[4] = (value >> 32) & 0xff;
-    ((char *)buf)[5] = (value >> 40) & 0xff;
-    ((char *)buf)[6] = (value >> 48) & 0xff;
-    ((char *)buf)[7] = (value >> 56) & 0xff;
+    ((uint8_t *)buf)[0] = value & 0xff;
+    ((uint8_t *)buf)[1] = (value >> 8) & 0xff;
+    ((uint8_t *)buf)[2] = (value >> 16) & 0xff;
+    ((uint8_t *)buf)[3] = (value >> 24) & 0xff;
+    ((uint8_t *)buf)[4] = (value >> 32) & 0xff;
+    ((uint8_t *)buf)[5] = (value >> 40) & 0xff;
+    ((uint8_t *)buf)[6] = (value >> 48) & 0xff;
+    ((uint8_t *)buf)[7] = (value >> 56) & 0xff;
   }
 
-  return POINTER_DRIFT(buf, sizeof(value));
+  return POINTER_SHIFT(buf, sizeof(value));
 }
 
 static FORCE_INLINE void *taosDecodeFixed16(void *buf, uint16_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
-    ((char *)value)[1] = ((char *)buf)[0];
-    ((char *)value)[0] = ((char *)buf)[1];
+    ((uint8_t *)value)[1] = ((uint8_t *)buf)[0];
+    ((uint8_t *)value)[0] = ((uint8_t *)buf)[1];
   }
 
-  return POINTER_DRIFT(buf, sizeof(*value));
+  return POINTER_SHIFT(buf, sizeof(*value));
 }
 
 static FORCE_INLINE void *taosDecodeFixed32(void *buf, uint32_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
-    ((char *)value)[3] = ((char *)buf)[0];
-    ((char *)value)[2] = ((char *)buf)[1];
-    ((char *)value)[1] = ((char *)buf)[2];
-    ((char *)value)[0] = ((char *)buf)[3];
+    ((uint8_t *)value)[3] = ((uint8_t *)buf)[0];
+    ((uint8_t *)value)[2] = ((uint8_t *)buf)[1];
+    ((uint8_t *)value)[1] = ((uint8_t *)buf)[2];
+    ((uint8_t *)value)[0] = ((uint8_t *)buf)[3];
   }
 
-  return POINTER_DRIFT(buf, sizeof(*value));
+  return POINTER_SHIFT(buf, sizeof(*value));
 }
 
 static FORCE_INLINE void *taosDecodeFixed64(void *buf, uint64_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
-    ((char *)value)[7] = ((char *)buf)[0];
-    ((char *)value)[6] = ((char *)buf)[1];
-    ((char *)value)[5] = ((char *)buf)[2];
-    ((char *)value)[4] = ((char *)buf)[3];
-    ((char *)value)[3] = ((char *)buf)[4];
-    ((char *)value)[2] = ((char *)buf)[5];
-    ((char *)value)[1] = ((char *)buf)[6];
-    ((char *)value)[0] = ((char *)buf)[7];
+    ((uint8_t *)value)[7] = ((uint8_t *)buf)[0];
+    ((uint8_t *)value)[6] = ((uint8_t *)buf)[1];
+    ((uint8_t *)value)[5] = ((uint8_t *)buf)[2];
+    ((uint8_t *)value)[4] = ((uint8_t *)buf)[3];
+    ((uint8_t *)value)[3] = ((uint8_t *)buf)[4];
+    ((uint8_t *)value)[2] = ((uint8_t *)buf)[5];
+    ((uint8_t *)value)[1] = ((uint8_t *)buf)[6];
+    ((uint8_t *)value)[0] = ((uint8_t *)buf)[7];
   }
 
-  return POINTER_DRIFT(buf, sizeof(*value));
+  return POINTER_SHIFT(buf, sizeof(*value));
 }
 
-// TODO
-static FORCE_INLINE void *taosEncodeVariant16(void *buf, uint16_t value) {}
-static FORCE_INLINE void *taosEncodeVariant32(void *buf, uint32_t value) {}
-static FORCE_INLINE void *taosEncodeVariant64(void *buf, uint64_t value) {}
-static FORCE_INLINE void *taosDecodeVariant16(void *buf, uint16_t *value) {}
-static FORCE_INLINE void *taosDecodeVariant32(void *buf, uint32_t *value) {}
-static FORCE_INLINE void *taosDecodeVariant64(void *buf, uint64_t *value) {}
+static FORCE_INLINE void *taosEncodeVariant16(void *buf, uint16_t value) {
+  int i = 0;
+  while (value >= ENCODE_LIMIT) {
+    ((uint8_t *)buf)[i] = (value | ENCODE_LIMIT);
+    value >>= 7;
+    i++;
+    ASSERT(i < 3);
+  }
+
+  ((uint8_t *)buf)[i] = value;
+
+  return POINTER_SHIFT(buf, i+1);
+}
+
+static FORCE_INLINE void *taosEncodeVariant32(void *buf, uint32_t value) {
+  int i = 0;
+  while (value >= ENCODE_LIMIT) {
+    ((uint8_t *)buf)[i] = (value | ENCODE_LIMIT);
+    value >>= 7;
+    i++;
+    ASSERT(i < 5);
+  }
+
+  ((uint8_t *)buf)[i] = value;
+
+  return POINTER_SHIFT(buf, i + 1);
+}
+
+static FORCE_INLINE void *taosEncodeVariant64(void *buf, uint64_t value) {
+  int i = 0;
+  while (value >= ENCODE_LIMIT) {
+    ((uint8_t *)buf)[i] = (value | ENCODE_LIMIT);
+    value >>= 7;
+    i++;
+    ASSERT(i < 10);
+  }
+
+  ((uint8_t *)buf)[i] = value;
+
+  return POINTER_SHIFT(buf, i + 1);
+}
+
+static FORCE_INLINE void *taosDecodeVariant16(void *buf, uint16_t *value) {
+  int i = 0;
+  uint16_t tval = 0;
+  *value = 0;
+  while (i < 3) {
+    tval = (uint16_t)(((uint8_t *)buf)[i]);
+    if (tval < ENCODE_LIMIT) {
+      (*value) |= (tval << (7 * i));
+      return POINTER_SHIFT(buf, i + 1);
+    } else {
+      (*value) |= ((tval & (ENCODE_LIMIT - 1)) << (7 * i));
+      i++;
+    }
+  }
+
+  return NULL;  // error happened
+}
+
+static FORCE_INLINE void *taosDecodeVariant32(void *buf, uint32_t *value) {
+  int i = 0;
+  uint32_t tval = 0;
+  *value = 0;
+  while (i < 5) {
+    tval = (uint32_t)(((uint8_t *)buf)[i]);
+    if (tval < ENCODE_LIMIT) {
+      (*value) |= (tval << (7 * i));
+      return POINTER_SHIFT(buf, i + 1);
+    } else {
+      (*value) |= ((tval & (ENCODE_LIMIT - 1)) << (7 * i));
+      i++;
+    }
+  }
+
+  return NULL;  // error happened
+}
+
+static FORCE_INLINE void *taosDecodeVariant64(void *buf, uint64_t *value) {
+  int i = 0;
+  uint64_t tval = 0;
+  *value = 0;
+  while (i < 10) {
+    tval = (uint64_t)(((uint8_t *)buf)[i]);
+    if (tval < ENCODE_LIMIT) {
+      (*value) |= (tval << (7 * i));
+      return POINTER_SHIFT(buf, i + 1);
+    } else {
+      (*value) |= ((tval & (ENCODE_LIMIT - 1)) << (7 * i));
+      i++;
+    }
+  }
+
+  return NULL;  // error happened
+}
 
 #ifdef __cplusplus
 }

src/util/inc/tutil.h

@@ -45,7 +45,7 @@ extern "C" {
 #define tclose(x) taosCloseSocket(x)
 
 // Pointer p drift right by b bytes
-#define POINTER_DRIFT(p, b) ((void *)((char *)(p) + (b)))
+#define POINTER_SHIFT(p, b) ((void *)((char *)(p) + (b)))
 
 #ifndef NDEBUG
 #define ASSERT(x) assert(x)

src/util/tests/codingTests.cpp

+#include <gtest/gtest.h>
+#include <stdlib.h>
+#include <time.h>
+#include <random>
+
+#include "tcoding.h"
+
+static bool test_fixed_uint16(uint16_t value) {
+  char     buf[20] = "\0";
+  uint16_t value_check = 0;
+
+  void *ptr1 = taosEncodeFixed16(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeFixed16(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+static bool test_fixed_uint32(uint32_t value) {
+  char     buf[20] = "\0";
+  uint32_t value_check = 0;
+
+  void *ptr1 = taosEncodeFixed32(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeFixed32(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+static bool test_fixed_uint64(uint64_t value) {
+  char     buf[20] = "\0";
+  uint64_t value_check = 0;
+
+  void *ptr1 = taosEncodeFixed64(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeFixed64(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+static bool test_variant_uint16(uint16_t value) {
+  char     buf[20] = "\0";
+  uint16_t value_check = 0;
+
+  void *ptr1 = taosEncodeVariant16(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeVariant16(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+static bool test_variant_uint32(uint32_t value) {
+  char     buf[20] = "\0";
+  uint32_t value_check = 0;
+
+  void *ptr1 = taosEncodeVariant32(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeVariant32(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+static bool test_variant_uint64(uint64_t value) {
+  char     buf[20] = "\0";
+  uint64_t value_check = 0;
+
+  void *ptr1 = taosEncodeVariant64(static_cast<void *>(buf), value);
+  void *ptr2 = taosDecodeVariant64(static_cast<void *>(buf), &value_check);
+
+  return ((ptr2 != NULL) && (value == value_check) && (ptr1 == ptr2));
+}
+
+TEST(codingTest, fixed_encode_decode) {
+  srand(time(0));
+
+  for (uint16_t value = 0; value <= UINT16_MAX; value++) {
+    ASSERT_TRUE(test_fixed_uint16(value));
+    if (value == UINT16_MAX) break;
+  }
+
+  ASSERT_TRUE(test_fixed_uint32(0));
+  ASSERT_TRUE(test_fixed_uint32(UINT32_MAX));
+
+  for (int i = 0; i < 1000000; i++) {
+    ASSERT_TRUE(test_fixed_uint32(rand()));
+  }
+
+  std::mt19937_64 gen (std::random_device{}());
+
+  ASSERT_TRUE(test_fixed_uint64(0));
+  ASSERT_TRUE(test_fixed_uint64(UINT64_MAX));
+  for (int i = 0; i < 1000000; i++) {
+    ASSERT_TRUE(test_fixed_uint64(gen()));
+  }
+}
+
+TEST(codingTest, variant_encode_decode) {
+  srand(time(0));
+
+  for (uint16_t value = 0; value <= UINT16_MAX; value++) {
+    ASSERT_TRUE(test_variant_uint16(value));
+    if (value == UINT16_MAX) break;
+  }
+
+  ASSERT_TRUE(test_variant_uint32(0));
+  ASSERT_TRUE(test_variant_uint32(UINT32_MAX));
+
+  for (int i = 0; i < 5000000; i++) {
+    ASSERT_TRUE(test_variant_uint32(rand()));
+  }
+
+  std::mt19937_64 gen (std::random_device{}());
+
+  ASSERT_TRUE(test_variant_uint64(0));
+  ASSERT_TRUE(test_variant_uint64(UINT64_MAX));
+  for (int i = 0; i < 5000000; i++) {
+    uint64_t value = gen();
+    // printf("%ull\n", value);
+    ASSERT_TRUE(test_variant_uint64(value));
+  }
+}
\ No newline at end of file