add/check tSma schema methods

include/common/tmsg.h

@@ -1866,10 +1866,59 @@ typedef struct {
   uint64_t  tableUid;  // super/common table uid
   int64_t   interval;
   int64_t   sliding;
-  col_id_t* colIds;   // N.B. sorted column ids
+  col_id_t* colIds;   // sorted column ids
-  uint16_t* funcIds;  // N.B. sorted sma function ids
+  uint16_t* funcIds;  // sorted sma function ids
 } STSma;              // Time-range-wise SMA
 
+typedef struct {
+  int8_t      msgType;  // 0 create, 1 recreate
+  STSma       tSma;
+  STimeWindow window;
+} SCreateTSmaMsg;
+
+typedef struct {
+  STimeWindow window;
+  char        indexName[TSDB_INDEX_NAME_LEN + 1];
+} SDropTSmaMsg;
+
+typedef struct {
+  STimeWindow tsWindow;     // [skey, ekey]
+  uint64_t    tableUid;     // sub/common table uid
+  int32_t     numOfBlocks;  // number of sma blocks for each column, total number is numOfBlocks*numOfColId
+  int32_t     dataLen;      // total data length
+  col_id_t*   colIds;       // e.g. 2,4,9,10
+  col_id_t    numOfColIds;  // e.g. 4
+  char        data[];       // the sma blocks
+} STSmaData;
+
+// TODO: move to the final location afte schema of STSma/STSmaData defined
+static FORCE_INLINE void tdDestroySmaData(STSmaData* pSmaData) {
+  if (pSmaData) {
+    if (pSmaData->colIds) {
+      tfree(pSmaData->colIds);
+    }
+    tfree(pSmaData);
+  }
+}
+
+// RSma: Time-range-wise Rollup SMA
+// TODO: refactor when rSma grammar defined finally =>
+typedef struct {
+  int64_t  interval;
+  int32_t  retention;  // unit: day
+  uint16_t days;       // unit: day
+  int8_t   intervalUnit;
+} SSmaParams;
+// TODO: refactor when rSma grammar defined finally <=
+
+typedef struct {
+  // TODO: refactor to use the real schema =>
+  STSma   tsma;
+  float   xFilesFactor;
+  SArray* smaParams;  // SSmaParams
+  // TODO: refactor to use the real schema <=
+} SRSma;
+
 typedef struct {
   uint32_t number;
   STSma*   tSma;
@@ -1885,12 +1934,17 @@ static FORCE_INLINE void tdDestroyTSma(STSma* pSma, bool releaseSelf) {
   }
 }
 
-static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW) {
+static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW, bool releaseSelf) {
-  if (pSW && pSW->tSma) {
+  if (pSW) {
-    for (uint32_t i = 0; i < pSW->number; ++i) {
+    if (pSW->tSma) {
-      tdDestroyTSma(pSW->tSma + i, false);
+      for (uint32_t i = 0; i < pSW->number; ++i) {
+        tdDestroyTSma(pSW->tSma + i, false);
+      }
+      tfree(pSW->tSma);
+    }
+    if (releaseSelf) {
+      free(pSW);
     }
-    tfree(pSW->tSma);
   }
 }
 

include/common/tmsgdef.h

@@ -184,6 +184,9 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
 
+  TD_DEF_MSG_TYPE(TDMT_VND_CREATE_SMA, "vnode-create-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_DROP_SMA, "vnode-drop-sma", NULL, NULL)
 
   // Requests handled by QNODE
   TD_NEW_MSG_SEG(TDMT_QND_MSG)

include/common/trow.h

@@ -118,6 +118,8 @@ typedef struct {
 } SKvRow;
 
 typedef struct {
+  /// timestamp
+  TSKEY ts;
   union {
     /// union field for encode and decode
     uint32_t info;
@@ -138,8 +140,6 @@ typedef struct {
   uint32_t len;
   /// row version
   uint64_t ver;
-  /// timestamp
-  TSKEY ts;
   /// the inline data, maybe a tuple or a k-v tuple
   char data[];
 } STSRow;
@@ -173,7 +173,7 @@ typedef struct {
 #define TD_ROW_DATA(r)     ((r)->data)
 #define TD_ROW_LEN(r)      ((r)->len)
 #define TD_ROW_KEY(r)      ((r)->ts)
-#define TD_ROW_KEY_ADDR(r) POINTER_SHIFT((r), 16)
+#define TD_ROW_KEY_ADDR(r) (r)
 
 // N.B. If without STSchema, getExtendedRowSize() is used to get the rowMaxBytes and
 // (int32_t)ceil((double)nCols/TD_VTYPE_PARTS) should be added if TD_SUPPORT_BITMAP defined.

source/dnode/vnode/inc/meta.h

@@ -57,6 +57,7 @@ STbCfg *        metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
 SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline);
 STSchema *      metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver);
 SSmaCfg *       metaGetSmaInfoByName(SMeta *pMeta, const char *indexName);
+STSmaWrapper *  metaGetSmaInfoByUid(SMeta *pMeta, tb_uid_t uid);
 
 SMTbCursor *metaOpenTbCursor(SMeta *pMeta);
 void        metaCloseTbCursor(SMTbCursor *pTbCur);

source/dnode/vnode/inc/tsdb.h

@@ -87,6 +87,27 @@ int    tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp);
 int    tsdbPrepareCommit(STsdb *pTsdb);
 int    tsdbCommit(STsdb *pTsdb);
 
+/**
+ * @brief Insert tSma(Time-range-wise SMA) data from stream computing engine
+ * 
+ * @param pTsdb 
+ * @param param 
+ * @param pData 
+ * @return int32_t 
+ */
+int32_t tsdbInsertTSmaData(STsdb *pTsdb, STSma *param, STSmaData *pData);
+
+/**
+ * @brief Insert RSma(Time-range-wise Rollup SMA) data.
+ * 
+ * @param pTsdb 
+ * @param param 
+ * @param pData 
+ * @return int32_t 
+ */
+int32_t tsdbInsertRSmaData(STsdb *pTsdb, SRSma *param, STSmaData *pData);
+
+
 // STsdbCfg
 int  tsdbOptionsInit(STsdbCfg *);
 void tsdbOptionsClear(STsdbCfg *);

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

@@ -35,6 +35,7 @@
 #include "tsdbMemory.h"
 #include "tsdbOptions.h"
 #include "tsdbReadImpl.h"
+#include "tsdbSma.h"
 
 #ifdef __cplusplus
 extern "C" {

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

@@ -42,6 +42,7 @@ typedef struct {
 typedef struct {
   STsdbFSMeta meta;  // FS meta
   SArray *    df;    // data file array
+  SArray *    smaf;  // sma data file array
 } SFSStatus;
 
 typedef struct {

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_TSDB_SMA_H_
+#define _TD_TSDB_SMA_H_
+
+// insert/update interface
+int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, STSma *param, STSmaData *pData);
+int32_t tsdbInsertRSmaDataImpl(STsdb *pTsdb, SRSma *param, STSmaData *pData);
+
+
+// query interface
+// TODO: This is the basic params, and should wrap the params to a queryHandle.
+int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, STSma *param, STSmaData *pData, STimeWindow *queryWin, int32_t nMaxResult);
+
+// management interface
+int32_t tsdbGetTSmaStatus(STsdb *pTsdb, STSma *param, void* result);
+int32_t tsdbRemoveTSmaData(STsdb *pTsdb, STSma *param, STimeWindow *pWin);
+
+
+
+
+// internal func
+static FORCE_INLINE int32_t tsdbEncodeTSmaKey(uint64_t tableUid, col_id_t colId, TSKEY tsKey, void **pData) {
+  int32_t len = 0;
+  len += taosEncodeFixedU64(pData, tableUid);
+  len += taosEncodeFixedU16(pData, colId);
+  len += taosEncodeFixedI64(pData, tsKey);
+  return len;
+}
+
+#if 0
+
+typedef struct {
+  int   minFid;
+  int   midFid;
+  int   maxFid;
+  TSKEY minKey;
+} SRtn;
+
+typedef struct {
+  uint64_t uid;
+  int64_t  offset;
+  int64_t  size;
+} SKVRecord;
+
+void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn);
+
+static FORCE_INLINE int TSDB_KEY_FID(TSKEY key, int32_t days, int8_t precision) {
+  if (key < 0) {
+    return (int)((key + 1) / tsTickPerDay[precision] / days - 1);
+  } else {
+    return (int)((key / tsTickPerDay[precision] / days));
+  }
+}
+
+static FORCE_INLINE int tsdbGetFidLevel(int fid, SRtn *pRtn) {
+  if (fid >= pRtn->maxFid) {
+    return 0;
+  } else if (fid >= pRtn->midFid) {
+    return 1;
+  } else if (fid >= pRtn->minFid) {
+    return 2;
+  } else {
+    return -1;
+  }
+}
+
+#define TSDB_DEFAULT_BLOCK_ROWS(maxRows) ((maxRows)*4 / 5)
+
+int   tsdbEncodeKVRecord(void **buf, SKVRecord *pRecord);
+void *tsdbDecodeKVRecord(void *buf, SKVRecord *pRecord);
+void *tsdbCommitData(STsdbRepo *pRepo);
+int   tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn);
+int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, SArray *pSubA, void **ppBuf, SBlockIdx *pIdx);
+int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
+int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock,
+                       bool isLast, bool isSuper, void **ppBuf, void **ppCBuf);
+int   tsdbApplyRtn(STsdbRepo *pRepo);
+
+#endif
+
+#endif /* _TD_TSDB_SMA_H_ */

source/dnode/vnode/src/meta/metaBDBImpl.c

@@ -833,6 +833,7 @@ SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) {
   }
 
   pCur->uid = uid;
+  // TODO: lock?
   ret = pDB->pCtbIdx->cursor(pDB->pSmaIdx, NULL, &(pCur->pCur), 0);
   if (ret != 0) {
     free(pCur);
@@ -852,25 +853,68 @@ void metaCloseSmaCurosr(SMSmaCursor *pCur) {
   }
 }
 
-const char* metaSmaCursorNext(SMSmaCursor *pCur) {
+const char *metaSmaCursorNext(SMSmaCursor *pCur) {
-  DBT    skey = {0};
+  DBT skey = {0};
-  DBT    pkey = {0};
+  DBT pkey = {0};
-  DBT    pval = {0};
+  DBT pval = {0};
-  void  *pBuf;
 
   // Set key
   skey.data = &(pCur->uid);
   skey.size = sizeof(pCur->uid);
-
+  // TODO: lock?
   if (pCur->pCur->pget(pCur->pCur, &skey, &pkey, &pval, DB_NEXT) == 0) {
-    const char* indexName = (const char *)pkey.data;
+    const char *indexName = (const char *)pkey.data;
     assert(indexName != NULL);
     return indexName;
   } else {
-    return 0;
+    return NULL;
   }
 }
 
+STSmaWrapper *metaGetSmaInfoByUid(SMeta *pMeta, tb_uid_t uid) {
+  STSmaWrapper *pSW = NULL;
+
+  pSW = calloc(sizeof(*pSW), 1);
+  if (pSW == NULL) {
+    return NULL;
+  }
+
+  SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid);
+  if (pCur == NULL) {
+    free(pSW);
+    return NULL;
+  }
+
+  DBT   skey = {.data = &(pCur->uid)};
+  DBT   pval = {.size = sizeof(pCur->uid)};
+  void *pBuf = NULL;
+
+  while (true) {
+    // TODO: lock?
+    if (pCur->pCur->pget(pCur->pCur, &skey, NULL, &pval, DB_NEXT) == 0) {
+      ++pSW->number;
+      STSma *tptr = (STSma *)realloc(pSW->tSma, pSW->number * sizeof(STSma));
+      if (tptr == NULL) {
+        metaCloseSmaCurosr(pCur);
+        tdDestroyTSmaWrapper(pSW, true);
+        return NULL;
+      }
+      pSW->tSma = tptr;
+      pBuf = pval.data;
+      if (tDecodeTSma(pBuf, pSW->tSma + pSW->number - 1) == NULL) {
+        metaCloseSmaCurosr(pCur);
+        tdDestroyTSmaWrapper(pSW, true);
+        return NULL;
+      }
+      continue;
+    }
+    break;
+  }
+
+  metaCloseSmaCurosr(pCur);
+
+  return pSW;
+}
 
 static void metaDBWLock(SMetaDB *pDB) {
 #if IMPL_WITH_LOCK

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

@@ -15,6 +15,14 @@
 
 #include "tsdbDef.h"
 
+/**
+ * @brief insert TS data
+ *
+ * @param pTsdb
+ * @param pMsg
+ * @param pRsp
+ * @return int
+ */
 int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
   // Check if mem is there. If not, create one.
   if (pTsdb->mem == NULL) {
@@ -24,4 +32,37 @@ int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
     }
   }
   return tsdbMemTableInsert(pTsdb, pTsdb->mem, pMsg, NULL);
+}
+
+/**
+ * @brief Insert/Update tSma(Time-range-wise SMA) data from stream computing engine
+ * 
+ * @param pTsdb 
+ * @param param 
+ * @param pData 
+ * @return int32_t 
+ * TODO: Who is responsible for resource allocate and release?
+ */
+int32_t tsdbInsertTSmaData(STsdb *pTsdb, STSma *param, STSmaData *pData) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if ((code = tsdbInsertTSmaDataImpl(pTsdb, param, pData)) < 0) {
+    tsdbWarn("vgId:%d insert tSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
+  }
+  return code;
+}
+
+/**
+ * @brief Insert Time-range-wise Rollup Sma(RSma) data
+ *
+ * @param pTsdb
+ * @param param
+ * @param pData
+ * @return int32_t
+ */
+int32_t tsdbInsertRSmaData(STsdb *pTsdb, SRSma *param, STSmaData *pData) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if ((code = tsdbInsertRSmaDataImpl(pTsdb, param, pData)) < 0) {
+    tsdbWarn("vgId:%d insert rSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
+  }
+  return code;
 }

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

@@ -132,6 +132,15 @@ int  vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       if (tqProcessRebReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
       }
     } break;
+    case TDMT_VND_CREATE_SMA: {  // timeRangeSMA
+      // 1. tdCreateSmaMeta(pVnode->pMeta,...);
+      // 2. tdCreateSmaDataInit();
+      // 3. tdCreateSmaData
+    } break;
+    case TDMT_VND_CANCEL_SMA: {  // timeRangeSMA
+    } break;
+    case TDMT_VND_DROP_SMA: {  // timeRangeSMA
+    } break;
     default:
       ASSERT(0);
       break;

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -95,7 +95,7 @@ TEST(testCase, tSmaEncodeDecodeTest) {
 
   // resource release
   tdDestroyTSma(&tSma, false);
-  tdDestroyTSmaWrapper(&dstTSmaWrapper);
+  tdDestroyTSmaWrapper(&dstTSmaWrapper, false);
 }
 
 TEST(testCase, tSma_DB_Put_Get_Del_Test) {
@@ -161,7 +161,7 @@ TEST(testCase, tSma_DB_Put_Get_Del_Test) {
   EXPECT_EQ(qSmaCfg->interval, tSma.interval);
   tdDestroyTSma(qSmaCfg, true);
 
-  // get value by table uid
+  // get index name by table uid
   SMSmaCursor *pSmaCur = metaOpenSmaCursor(pMeta, tbUid);
   assert(pSmaCur != NULL);
   uint32_t indexCnt = 0;
@@ -176,6 +176,15 @@ TEST(testCase, tSma_DB_Put_Get_Del_Test) {
   EXPECT_EQ(indexCnt, 2);
   metaCloseSmaCurosr(pSmaCur);
 
+  // get wrapper by table uid
+  STSmaWrapper *pSW = metaGetSmaInfoByUid(pMeta, tbUid);
+  assert(pSW != NULL);
+  EXPECT_EQ(pSW->number, 2);
+  EXPECT_STRCASEEQ(pSW->tSma->indexName, smaIndexName1);
+  EXPECT_EQ(pSW->tSma->tableUid, tSma.tableUid);
+  EXPECT_STRCASEEQ((pSW->tSma + 1)->indexName, smaIndexName2);
+  EXPECT_EQ((pSW->tSma + 1)->tableUid, tSma.tableUid);
+
   // resource release
   metaRemoveSmaFromDb(pMeta, smaIndexName1);
   metaRemoveSmaFromDb(pMeta, smaIndexName2);
@@ -197,15 +206,15 @@ TEST(testCase, tSmaInsertTest) {
 
   int32_t blockSize = tSma.numOfFuncIds * sizeof(int64_t);
   int32_t numOfColIds = 3;
-  int32_t numOfSmaBlocks = 10;
+  int32_t numOfBlocks = 10;
 
-  int32_t dataLen = numOfColIds * numOfSmaBlocks * blockSize;
+  int32_t dataLen = numOfColIds * numOfBlocks * blockSize;
 
   pSmaData = (STSmaData*)malloc(sizeof(STSmaData) + dataLen);
   ASSERT_EQ(pSmaData != NULL, true);
   pSmaData->tableUid = 3232329230;
   pSmaData->numOfColIds = numOfColIds;
-  pSmaData->numOfSmaBlocks = numOfSmaBlocks;
+  pSmaData->numOfBlocks = numOfBlocks;
   pSmaData->dataLen = dataLen;
   pSmaData->tsWindow.skey = 1640000000;
   pSmaData->tsWindow.ekey = 1645788649;