Merge pull request #12092 from taosdata/feature/TD-14481-3.0

feat: rollup sma data query

source/dnode/vnode/src/inc/tsdb.h

@@ -186,6 +186,7 @@ struct STsdbFS {
 
 #define REPO_ID(r)        TD_VID((r)->pVnode)
 #define REPO_CFG(r)       (&(r)->pVnode->config.tsdbCfg)
+#define REPO_LEVEL(r)     ((r)->level)
 #define REPO_FS(r)        ((r)->fs)
 #define REPO_META(r)      ((r)->pVnode->pMeta)
 #define REPO_TFS(r)       ((r)->pVnode->pTfs)

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -15,6 +15,8 @@
 
 #include "tsdb.h"
 
+extern const char *TSDB_LEVEL_DNAME[];
+
 typedef enum { TSDB_TXN_TEMP_FILE = 0, TSDB_TXN_CURR_FILE } TSDB_TXN_FILE_T;
 static const char *tsdbTxnFname[] = {"current.t", "current"};
 #define TSDB_MAX_FSETS(keep, days) ((keep) / (days) + 3)
@@ -35,12 +37,12 @@ static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired);
 // static int  tsdbProcessExpiredFS(STsdb *pRepo);
 // static int  tsdbCreateMeta(STsdb *pRepo);
 
-static void tsdbGetRootDir(int repoid, char dirName[]) {
+static void tsdbGetRootDir(int repoid, int8_t level, char dirName[]) {
-  snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/tsdb", repoid);
+  snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/%s", repoid, TSDB_LEVEL_DNAME[level]);
 }
 
-static void tsdbGetDataDir(int repoid, char dirName[]) {
+static void tsdbGetDataDir(int repoid, int8_t level, char dirName[]) {
-  snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/tsdb/data", repoid);
+  snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/%s/data", repoid, TSDB_LEVEL_DNAME[level]);
 }
 
 // For backward compatibility
@@ -588,8 +590,8 @@ static int tsdbComparFidFSet(const void *arg1, const void *arg2) {
 }
 
 static void tsdbGetTxnFname(STsdb *pRepo, TSDB_TXN_FILE_T ftype, char fname[]) {
-  snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/tsdb/%s", tfsGetPrimaryPath(REPO_TFS(pRepo)), REPO_ID(pRepo),
+  snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/%s/%s", tfsGetPrimaryPath(REPO_TFS(pRepo)), REPO_ID(pRepo),
-           tsdbTxnFname[ftype]);
+           TSDB_LEVEL_DNAME[REPO_LEVEL(pRepo)], tsdbTxnFname[ftype]);
 }
 
 static int tsdbOpenFSFromCurrent(STsdb *pRepo) {
@@ -719,7 +721,7 @@ static int tsdbScanRootDir(STsdb *pRepo) {
   STsdbFS        *pfs = REPO_FS(pRepo);
   const STfsFile *pf;
 
-  tsdbGetRootDir(REPO_ID(pRepo), rootDir);
+  tsdbGetRootDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), rootDir);
   STfsDir *tdir = tfsOpendir(REPO_TFS(pRepo), rootDir);
   if (tdir == NULL) {
     tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno));
@@ -753,7 +755,7 @@ static int tsdbScanDataDir(STsdb *pRepo) {
   STsdbFS        *pfs = REPO_FS(pRepo);
   const STfsFile *pf;
 
-  tsdbGetDataDir(REPO_ID(pRepo), dataDir);
+  tsdbGetDataDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), dataDir);
   STfsDir *tdir = tfsOpendir(REPO_TFS(pRepo), dataDir);
   if (tdir == NULL) {
     tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno));
@@ -801,7 +803,7 @@ static int tsdbRestoreDFileSet(STsdb *pRepo) {
   regex_t         regex;
   STsdbFS        *pfs = REPO_FS(pRepo);
 
-  tsdbGetDataDir(REPO_ID(pRepo), dataDir);
+  tsdbGetDataDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), dataDir);
 
   // Resource allocation and init
   regcomp(&regex, pattern, REG_EXTENDED);

source/dnode/vnode/src/tsdb/tsdbFile.c

@@ -27,7 +27,7 @@ static const char *TSDB_FNAME_SUFFIX[] = {
     "rsma",  // TSDB_FILE_RSMA
 };
 
-static const char *TSDB_LEVEL_DNAME[] = {
+const char *TSDB_LEVEL_DNAME[] = {
     "tsdb",
     "rsma1",
     "rsma2",

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -164,6 +164,8 @@ static int32_t tsdbCheckInfoCompar(const void* key1, const void* key2);
 // static void*   destroyTableCheckInfo(SArray* pTableCheckInfo);
 static bool tsdbGetExternalRow(tsdbReaderT pHandle);
 
+static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions);
+
 static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
   pBlockLoadInfo->slot = -1;
   pBlockLoadInfo->uid = 0;
@@ -350,12 +352,38 @@ static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableData
               pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
   }
 }
-
+#if 0
+int           nQUERY = 0;
+#endif
 static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions) {
   if (vnodeIsRollup(pVnode)) {
-    // for(int32_t i=0; i< TSDB_; ) {
+    int level = 0;
-      
+#if 1
-    // }
+    int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
+    for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
+      SRetention* pRetention = retentions + i;
+      if (pRetention->keep <= 0 || (now - pRetention->keep) >= winSKey) {
+        break;
+      }
+    }
+#endif
+#if 0
+    ++nQUERY;
+    if(nQUERY%3 == 0) {
+      level = 2;
+    } else if(nQUERY%2 == 0) {
+      level = 1;
+    } else {
+      level = 0;
+    }
+#endif
+    if (level == TSDB_RETENTION_L0) {
+      return VND_RSMA0(pVnode);
+    } else if (level == TSDB_RETENTION_L1) {
+      return VND_RSMA1(pVnode);
+    } else {
+      return VND_RSMA2(pVnode);
+    }
   }
   return pVnode->pTsdb;
 }
@@ -420,8 +448,10 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
     }
 
     pReadHandle->suppInfo.defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true);
-    pReadHandle->suppInfo.slotIds = taosMemoryMalloc(sizeof(int32_t) * taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn));
+    pReadHandle->suppInfo.slotIds =
-    pReadHandle->suppInfo.plist = taosMemoryCalloc(taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn), POINTER_BYTES);
+        taosMemoryMalloc(sizeof(int32_t) * taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn));
+    pReadHandle->suppInfo.plist =
+        taosMemoryCalloc(taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn), POINTER_BYTES);
   }
 
   pReadHandle->pDataCols = tdNewDataCols(1000, pVnode->config.tsdbCfg.maxRows);
@@ -444,7 +474,6 @@ _end:
 
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
                              uint64_t taskId) {
-
   STsdbReadHandle* pTsdbReadHandle = tsdbQueryTablesImpl(pVnode, pCond, qId, taskId);
   if (pTsdbReadHandle == NULL) {
     return NULL;

source/dnode/vnode/src/vnd/vnodeCommit.c

@@ -229,10 +229,28 @@ int vnodeCommit(SVnode *pVnode) {
     ASSERT(0);
     return -1;
   }
+
+  if(vnodeIsRollup(pVnode)) {
+    if (tsdbCommit(VND_RSMA0(pVnode)) < 0) {
+      ASSERT(0);
+      return -1;
+    }
+    if (tsdbCommit(VND_RSMA1(pVnode)) < 0) {
+      ASSERT(0);
+      return -1;
+    }
+    if (tsdbCommit(VND_RSMA2(pVnode)) < 0) {
+      ASSERT(0);
+      return -1;
+    }
+  } else {
     if (tsdbCommit(pVnode->pTsdb) < 0) {
       ASSERT(0);
       return -1;
     }
+  }
+
+  
   if (tqCommit(pVnode->pTq) < 0) {
     ASSERT(0);
     return -1;