TD-185

src/tsdb/inc/tsdbMain.h

@@ -210,12 +210,17 @@ typedef enum {
 extern const char *tsdbFileSuffix[];
 
 typedef struct {
-  int64_t size;      // total size of the file
+  uint32_t offset;
-  int64_t tombSize;  // unused file size
+  uint32_t len;
-  int32_t totalBlocks;
+  uint64_t size;      // total size of the file
-  int32_t totalSubBlocks;
+  uint64_t tombSize;  // unused file size
+  uint32_t totalBlocks;
+  uint32_t totalSubBlocks;
 } SFileInfo;
 
+void *tsdbEncodeSFileInfo(void *buf, const SFileInfo *pInfo);
+void *tsdbDecodeSFileInfo(void *buf, SFileInfo *pInfo);
+
 typedef struct {
   int       fd;
   char      fname[128];
@@ -242,8 +247,7 @@ typedef struct {
 
 STsdbFileH *tsdbInitFileH(char *dataDir, STsdbCfg *pCfg);
 void        tsdbCloseFileH(STsdbFileH *pFileH);
-int         tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, SFile *pFile, int writeHeader,
+int         tsdbCreateFile(char *dataDir, int fileId, const char *suffix, SFile *pFile);
-                           int toClose);
 SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables);
 int         tsdbOpenFile(SFile *pFile, int oflag);
 int         tsdbCloseFile(SFile *pFile);
@@ -266,15 +270,18 @@ void        tsdbSeekFileGroupIter(SFileGroupIter *pIter, int fid);
 SFileGroup *tsdbGetFileGroupNext(SFileGroupIter *pIter);
 
 typedef struct {
-  int32_t len;
+  uint32_t len;
-  int32_t offset;
+  uint32_t offset;
-  int32_t padding; // For padding purpose
+  uint32_t padding;  // For padding purpose
-  int32_t hasLast : 1;
+  uint32_t hasLast : 2;
-  int32_t numOfBlocks : 31;
+  uint32_t numOfBlocks : 30;
   int64_t uid;
   TSKEY   maxKey;
 } SCompIdx; /* sizeof(SCompIdx) = 28 */
 
+void *tsdbEncodeSCompIdx(void *buf, SCompIdx *pIdx);
+void *tsdbDecodeSCompIdx(void *buf, SCompIdx *pIdx);
+
 /**
  * if numOfSubBlocks == 0, then the SCompBlock is a sub-block
  * if numOfSubBlocks >= 1, then the SCompBlock is a super-block

src/tsdb/src/tsdbFile.c

@@ -38,7 +38,6 @@ const char *tsdbFileSuffix[] = {
 static int compFGroupKey(const void *key, const void *fgroup);
 static int compFGroup(const void *arg1, const void *arg2);
 static int tsdbWriteFileHead(SFile *pFile);
-static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables);
 static int tsdbOpenFGroup(STsdbFileH *pFileH, char *dataDir, int fid);
 
 STsdbFileH *tsdbInitFileH(char *dataDir, STsdbCfg *pCfg) {
@@ -121,8 +120,7 @@ SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int max
   if (pGroup == NULL) {  // if not exists, create one
     pFGroup->fileId = fid;
     for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
-      if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], maxTables, &(pFGroup->files[type]),
+      if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], &(pFGroup->files[type])) < 0)
-                         type == TSDB_FILE_TYPE_HEAD ? 1 : 0, 1) < 0)
         goto _err;
     }
 
@@ -298,29 +296,6 @@ static int tsdbWriteFileHead(SFile *pFile) {
   return 0;
 }
 
-static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables) {
-  int   size = sizeof(SCompIdx) * maxTables + sizeof(TSCKSUM);
-  void *buf = calloc(1, size);
-  if (buf == NULL) return -1;
-
-  if (lseek(pFile->fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) {
-    free(buf);
-    return -1;
-  }
-
-  taosCalcChecksumAppend(0, (uint8_t *)buf, size);
-
-  if (write(pFile->fd, buf, size) < 0) {
-    free(buf);
-    return -1;
-  }
-
-  pFile->info.size += size;
-
-  free(buf);
-  return 0;
-}
-
 int tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname) {
   if (dataDir == NULL || fname == NULL) return -1;
 
@@ -354,7 +329,7 @@ SFileGroup * tsdbOpenFilesForCommit(STsdbFileH *pFileH, int fid) {
   return pGroup;
 }
 
-int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, SFile *pFile, int writeHeader, int toClose) {
+int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, SFile *pFile) {
   memset((void *)pFile, 0, sizeof(SFile));
   pFile->fd = -1;
 
@@ -370,19 +345,12 @@ int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables,
     return -1;
   }
 
-  if (writeHeader) {
-    if (tsdbWriteHeadFileIdx(pFile, maxTables) < 0) {
-      tsdbCloseFile(pFile);
-      return -1;
-    }
-  }
-
   if (tsdbWriteFileHead(pFile) < 0) {
     tsdbCloseFile(pFile);
     return -1;
   }
 
-  if (toClose) tsdbCloseFile(pFile);
+  tsdbCloseFile(pFile);
 
   return 0;
 }

src/tsdb/src/tsdbMain.c

@@ -1148,8 +1148,16 @@ static void tsdbAlterKeep(STsdbRepo *pRepo, int32_t keep) {
 }
 
 static void tsdbAlterMaxTables(STsdbRepo *pRepo, int32_t maxTables) {
-  // TODO
   int oldMaxTables = pRepo->config.maxTables;
+  if (oldMaxTables < pRepo->config.maxTables) {
+    // TODO
+  }
+
+  STsdbMeta *pMeta = pRepo->tsdbMeta;
+
+  pMeta->maxTables = maxTables;
+  pMeta->tables = realloc(pMeta->tables, maxTables * sizeof(STable *));
+
   tsdbTrace("vgId:%d, tsdb maxTables is changed from %d to %d!", pRepo->config.tsdbId, oldMaxTables, maxTables);
 }
 

src/tsdb/src/tsdbMeta.c

@@ -154,7 +154,6 @@ STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables) {
   STsdbMeta *pMeta = (STsdbMeta *)malloc(sizeof(STsdbMeta));
   if (pMeta == NULL) return NULL;
 
-  pMeta->maxTables = maxTables;
   pMeta->nTables = 0;
   pMeta->superList = NULL;
   pMeta->tables = (STable **)calloc(maxTables, sizeof(STable *));

src/tsdb/src/tsdbRWHelper.c

@@ -18,6 +18,7 @@
 #include "tchecksum.h"
 #include "tscompression.h"
 #include "talgo.h"
+#include "tcoding.h"
 
 // Local function definitions
 // static int  tsdbCheckHelperCfg(SHelperCfg *pCfg);
@@ -211,13 +212,15 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
 
     // Create and open .h
     if (tsdbOpenFile(&(pHelper->files.nHeadF), O_WRONLY | O_CREAT) < 0) return -1;
-    size_t tsize = TSDB_FILE_HEAD_SIZE + sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM);
+    // size_t tsize = TSDB_FILE_HEAD_SIZE + sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM);
-    if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, tsize) < tsize) goto _err;
+    if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE)
+      goto _err;
 
     // Create and open .l file if should
     if (tsdbShouldCreateNewLast(pHelper)) {
       if (tsdbOpenFile(&(pHelper->files.nLastF), O_WRONLY | O_CREAT) < 0) goto _err;
-      if (tsendfile(pHelper->files.nLastF.fd, pHelper->files.lastF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) goto _err;
+      if (tsendfile(pHelper->files.nLastF.fd, pHelper->files.lastF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE)
+        goto _err;
     }
   } else {
     if (tsdbOpenFile(&(pHelper->files.dataF), O_RDONLY) < 0) goto _err;
@@ -426,13 +429,26 @@ int tsdbWriteCompInfo(SRWHelper *pHelper) {
 
 int tsdbWriteCompIdx(SRWHelper *pHelper) {
   ASSERT(TSDB_HELPER_TYPE(pHelper) == TSDB_WRITE_HELPER);
-  if (lseek(pHelper->files.nHeadF.fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1;
+  off_t offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END);
+  if (offset < 0) return -1;
 
-  ASSERT(tsizeof(pHelper->pCompIdx) == sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM));
+  SFile *pFile = &(pHelper->files.nHeadF);
-  taosCalcChecksumAppend(0, (uint8_t *)pHelper->pCompIdx, tsizeof(pHelper->pCompIdx));
+  pFile->info.offset = offset;
 
-  if (twrite(pHelper->files.nHeadF.fd, (void *)pHelper->pCompIdx, tsizeof(pHelper->pCompIdx)) < tsizeof(pHelper->pCompIdx))
+  void *buf = pHelper->blockBuffer;
-    return -1;
+  for (uint32_t i = 0; i < pHelper->config.maxTables; i++) {
+    SCompIdx *pCompIdx = pHelper->pCompIdx + i;
+    if (pCompIdx->offset > 0) {
+      buf = taosEncodeVariant32(buf, i);
+      buf = tsdbEncodeSCompIdx(buf, pCompIdx);
+    }
+  }
+
+  int tsize = (char *)buf - (char *)pHelper->blockBuffer + sizeof(TSCKSUM);
+  taosCalcChecksumAppend(0, (uint8_t *)pHelper->blockBuffer, tsize);
+
+  if (twrite(pHelper->files.nHeadF.fd, (void *)pHelper->blockBuffer, tsize) < tsize) return -1;
+  pFile->info.len = tsize;
   return 0;
 }
 
@@ -441,14 +457,34 @@ int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) {
 
   if (!helperHasState(pHelper, TSDB_HELPER_IDX_LOAD)) {
     // If not load from file, just load it in object
-    int fd = pHelper->files.headF.fd;
+    SFile *pFile = &(pHelper->files.headF);
+    int fd = pFile->fd;
 
-    if (lseek(fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1;
+    memset(pHelper->pCompIdx, 0, tsizeof(pHelper->pCompIdx));
-    if (tread(fd, (void *)(pHelper->pCompIdx), tsizeof((void *)pHelper->pCompIdx)) < tsizeof(pHelper->pCompIdx)) return -1;
+    if (pFile->info.offset > 0) {
-    if (!taosCheckChecksumWhole((uint8_t *)(pHelper->pCompIdx), tsizeof((void *)pHelper->pCompIdx))) {
+      ASSERT(pFile->info.offset > TSDB_FILE_HEAD_SIZE);
-      // TODO: File is broken, try to deal with it
+
-      return -1;
+      if (lseek(fd, pFile->info.offset, SEEK_SET) < 0) return -1;
+      if (tread(fd, (void *)(pHelper->blockBuffer), pFile->info.len) < pFile->info.len)
+        return -1;
+      if (!taosCheckChecksumWhole((uint8_t *)(pHelper->blockBuffer), pFile->info.len)) {
+        // TODO: File is broken, try to deal with it
+        return -1;
+      }
+
+      // Decode it
+      void *ptr = pHelper->blockBuffer;
+      while ((char *)ptr - (char *)pHelper->blockBuffer >= pFile->info.len - sizeof(TSCKSUM)) {
+        uint32_t tid = 0;
+        if ((ptr = taosDecodeVariant32(ptr, &tid)) == NULL) return -1;
+        ASSERT(tid > 0 && tid < pHelper->config.maxTables);
+
+        if ((ptr = tsdbDecodeSCompIdx(ptr, pHelper->pCompIdx + tid)) == NULL) return -1;
+
+        ASSERT((char *)ptr - (char *)pHelper->blockBuffer <= pFile->info.len - sizeof(TSCKSUM));
+      }
     }
+
   }
   helperSetState(pHelper, TSDB_HELPER_IDX_LOAD);
 
@@ -1169,3 +1205,75 @@ static int tsdbGetRowsInRange(SDataCols *pDataCols, TSKEY minKey, TSKEY maxKey)
 
   return ((TSKEY *)ptr2 - (TSKEY *)ptr1) + 1;
 }
+
+void *tsdbEncodeSCompIdx(void *buf, SCompIdx *pIdx) {
+  buf = taosEncodeVariant32(buf, pIdx->len);
+  buf = taosEncodeVariant32(buf, pIdx->offset);
+  buf = taosEncodeFixed8(buf, pIdx->hasLast);
+  buf = taosEncodeVariant32(buf, pIdx->numOfBlocks);
+  buf = taosEncodeFixed64(buf, pIdx->uid);
+  buf = taosEncodeFixed64(buf, pIdx->maxKey);
+
+  return buf;
+}
+
+void *tsdbDecodeSCompIdx(void *buf, SCompIdx *pIdx) {
+  uint8_t  hasLast = 0;
+  uint32_t numOfBlocks = 0;
+  uint64_t value = 0;
+
+  if ((buf = taosDecodeVariant32(buf, &(pIdx->len))) == NULL) return NULL;
+  if ((buf = taosDecodeVariant32(buf, &(pIdx->offset))) == NULL) return NULL;
+  if ((buf = taosDecodeFixed8(buf, &(hasLast))) == NULL) return NULL;
+  pIdx->hasLast = hasLast;
+  if ((buf = taosDecodeVariant32(buf, &(numOfBlocks))) == NULL) return NULL;
+  pIdx->numOfBlocks = numOfBlocks;
+  if ((buf = taosDecodeFixed64(buf, &value)) == NULL) return NULL;
+  pIdx->uid = (int64_t)value;
+  if ((buf = taosDecodeFixed64(buf, &value)) == NULL) return NULL;
+  pIdx->maxKey = (TSKEY)value;
+
+  return buf;
+}
+
+int tsdbUpdateFileHeader(SFile *pFile, uint32_t version) {
+  char buf[TSDB_FILE_HEAD_SIZE] = "\0";
+
+  void *pBuf = (void *)buf;
+  pBuf = taosEncodeFixed32(pBuf, version);
+  pBuf = tsdbEncodeSFileInfo(pBuf, &(pFile->info));
+  int tsize = (char *)pBuf - buf + sizeof(TSCKSUM);
+
+  ASSERT(tsize <= TSDB_FILE_HEAD_SIZE);
+
+  taosCalcChecksumAppend(0, (uint8_t *)buf, tsize);
+
+  if (lseek(pFile->fd, 0, SEEK_SET) < 0) return -1;
+  if (twrite(pFile->fd, (void *)buf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) return -1;
+
+  return 0;
+}
+
+
+
+void *tsdbEncodeSFileInfo(void *buf, const SFileInfo *pInfo) {
+  buf = taosEncodeFixed32(buf, pInfo->offset);
+  buf = taosEncodeFixed32(buf, pInfo->len);
+  buf = taosEncodeFixed64(buf, pInfo->size);
+  buf = taosEncodeFixed64(buf, pInfo->tombSize);
+  buf = taosEncodeFixed32(buf, pInfo->totalBlocks);
+  buf = taosEncodeFixed32(buf, pInfo->totalSubBlocks);
+
+  return buf;
+}
+
+void *tsdbDecodeSFileInfo(void *buf, SFileInfo *pInfo) {
+  buf = taosDecodeFixed32(buf, &(pInfo->offset));
+  buf = taosDecodeFixed32(buf, &(pInfo->len));
+  buf = taosDecodeFixed64(buf, &(pInfo->size));
+  buf = taosDecodeFixed64(buf, &(pInfo->tombSize));
+  buf = taosDecodeFixed32(buf, &(pInfo->totalBlocks));
+  buf = taosDecodeFixed32(buf, &(pInfo->totalSubBlocks));
+
+  return buf;
+}

src/util/inc/tcoding.h

@@ -29,6 +29,11 @@ extern "C" {
 static const int32_t TNUMBER = 1;
 #define IS_LITTLE_ENDIAN() (*(uint8_t *)(&TNUMBER) != 0)
 
+static FORCE_INLINE void *taosEncodeFixed8(void *buf, uint8_t value) {
+  ((uint8_t *)buf)[0] = value;
+  return POINTER_SHIFT(buf, sizeof(value));
+}
+
 static FORCE_INLINE void *taosEncodeFixed16(void *buf, uint16_t value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(buf, &value, sizeof(value));
@@ -70,6 +75,11 @@ static FORCE_INLINE void *taosEncodeFixed64(void *buf, uint64_t value) {
   return POINTER_SHIFT(buf, sizeof(value));
 }
 
+static FORCE_INLINE void *taosDecodeFixed8(void *buf, uint8_t *value) {
+  *value = ((uint8_t *)buf)[0];
+  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));