Merge branch '3.0' of github.com:taosdata/TDengine into feature/udf

include/common/tmsg.h

@@ -252,6 +252,7 @@ STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
 
 typedef struct {
+  int32_t code;
   int8_t  hashMeta;
   int64_t uid;
   char*   tblFName;
@@ -271,7 +272,7 @@ typedef struct {
 
 int32_t tEncodeSSubmitRsp(SEncoder* pEncoder, const SSubmitRsp* pRsp);
 int32_t tDecodeSSubmitRsp(SDecoder* pDecoder, SSubmitRsp* pRsp);
-void tFreeSSubmitRsp(SSubmitRsp *pRsp);
+void    tFreeSSubmitRsp(SSubmitRsp* pRsp);
 
 #define COL_SMA_ON  ((int8_t)0x1)
 #define COL_IDX_ON  ((int8_t)0x2)

source/common/src/tmsg.c

@@ -4032,6 +4032,7 @@ int32_t tDecodeSVSubmitReq(SDecoder *pCoder, SVSubmitReq *pReq) {
 static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBlock) {
   if (tStartEncode(pEncoder) < 0) return -1;
 
+  if (tEncodeI32(pEncoder, pBlock->code) < 0) return -1;
   if (tEncodeI8(pEncoder, pBlock->hashMeta) < 0) return -1;
   if (pBlock->hashMeta) {
     if (tEncodeI64(pEncoder, pBlock->uid) < 0) return -1;
@@ -4047,10 +4048,11 @@ static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBl
 static int32_t tDecodeSSubmitBlkRsp(SDecoder *pDecoder, SSubmitBlkRsp *pBlock) {
   if (tStartDecode(pDecoder) < 0) return -1;
 
+  if (tDecodeI32(pDecoder, &pBlock->code) < 0) return -1;
   if (tDecodeI8(pDecoder, &pBlock->hashMeta) < 0) return -1;
   if (pBlock->hashMeta) {
     if (tDecodeI64(pDecoder, &pBlock->uid) < 0) return -1;
-    pBlock->tblFName= taosMemoryCalloc(TSDB_TABLE_FNAME_LEN, 1);
+    pBlock->tblFName = taosMemoryCalloc(TSDB_TABLE_FNAME_LEN, 1);
     if (NULL == pBlock->tblFName) return -1;
     if (tDecodeCStrTo(pDecoder, pBlock->tblFName) < 0) return -1;
   }
@@ -4089,7 +4091,7 @@ int32_t tDecodeSSubmitRsp(SDecoder *pDecoder, SSubmitRsp *pRsp) {
     if (tDecodeSSubmitBlkRsp(pDecoder, pRsp->pBlocks + iBlock) < 0) return -1;
   }
 
-  tEndDecode(pDecoder);  
+  tEndDecode(pDecoder);
   tDecoderClear(pDecoder);
   return 0;
 }
@@ -4108,4 +4110,3 @@ void tFreeSSubmitRsp(SSubmitRsp *pRsp) {
 
   taosMemoryFree(pRsp);
 }
-

source/dnode/mnode/impl/src/mndStb.c

@@ -318,6 +318,7 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
   pOld->updateTime = pNew->updateTime;
   pOld->version = pNew->version;
   pOld->nextColId = pNew->nextColId;
+  pOld->ttl = pNew->ttl;
   pOld->numOfColumns = pNew->numOfColumns;
   pOld->numOfTags = pNew->numOfTags;
   memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));
@@ -832,7 +833,7 @@ static int32_t mndProcessVCreateStbRsp(SNodeMsg *pRsp) {
 }
 
 static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {
-  if (pAlter->commentLen != 0) return 0;
+  if (pAlter->commentLen != 0 || pAlter->ttl != 0) return 0;
 
   if (pAlter->numOfFields < 1 || pAlter->numOfFields != (int32_t)taosArrayGetSize(pAlter->pFields)) {
     terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
@@ -883,7 +884,8 @@ static int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) {
   return 0;
 }
 
-static int32_t mndUpdateStbComment(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen) {
+static int32_t mndUpdateStbCommentAndTTL(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen,
+                                         int32_t ttl) {
   if (commentLen > 0) {
     pNew->commentLen = commentLen;
     pNew->comment = taosMemoryCalloc(1, commentLen);
@@ -893,6 +895,9 @@ static int32_t mndUpdateStbComment(const SStbObj *pOld, SStbObj *pNew, char *pCo
     }
     memcpy(pNew->comment, pComment, commentLen);
   }
+  if (ttl >= 0) {
+    pNew->ttl = ttl;
+  }
 
   if (mndAllocStbSchemas(pOld, pNew) != 0) {
     return -1;
@@ -1232,7 +1237,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SNodeMsg *pReq, const SMAlterStbReq *
       code = mndAlterStbColumnBytes(pOld, &stbObj, pField0);
       break;
     case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
-      code = mndUpdateStbComment(pOld, &stbObj, pAlter->comment, pAlter->commentLen);
+      code = mndUpdateStbCommentAndTTL(pOld, &stbObj, pAlter->comment, pAlter->commentLen, pAlter->ttl);
       break;
     default:
       terrno = TSDB_CODE_OPS_NOT_SUPPORT;
@@ -1723,7 +1728,7 @@ static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlo
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)&pStb->updateTime, false);  // number of tables
 
-    char *p = taosMemoryMalloc(pStb->commentLen + VARSTR_HEADER_SIZE);  // check malloc failures
+    char *p = taosMemoryCalloc(1, pStb->commentLen + 1 + VARSTR_HEADER_SIZE);  // check malloc failures
     if (p != NULL) {
       if (pStb->commentLen != 0) {
         STR_TO_VARSTR(p, pStb->comment);

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

@@ -310,6 +310,17 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
   TSKEY          keyMax;
   SSubmitBlk    *pBlkCopy;
 
+  // check if table exists
+  SMetaReader mr = {0};
+  SMetaEntry  me = {0};
+  metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0);
+  if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) {
+    metaReaderClear(&mr);
+    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
+    return -1;
+  }
+  metaReaderClear(&mr);
+
   // create container is nedd
   tptr = taosHashGet(pMemTable->pHashIdx, &(pMsgIter->uid), sizeof(pMsgIter->uid));
   if (tptr == NULL) {

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

@@ -502,7 +502,7 @@ _exit:
   return 0;
 }
 
-static int vnodeDebugPrintSingleSubmitMsg(SMeta         *pMeta, SSubmitBlk  *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
+static int vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
   SSubmitBlkIter blkIter = {0};
   STSchema      *pSchema = NULL;
   tb_uid_t       suid = 0;
@@ -544,7 +544,7 @@ static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char
   while (true) {
     if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
     if (pBlock == NULL) break;
-    
+
     vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
   }
 
@@ -595,7 +595,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
       if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
         if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
-          pRsp->code = terrno;
+          submitBlkRsp.code = terrno;
           tDecoderClear(&decoder);
           goto _exit;
         }
@@ -617,8 +617,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
     }
 
     if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &submitBlkRsp) < 0) {
-      pRsp->code = terrno;
+      submitBlkRsp.code = terrno;
-      goto _exit;
     }
 
     submitRsp.numOfRows += submitBlkRsp.numOfRows;

source/libs/function/inc/builtinsimpl.h

@@ -95,6 +95,9 @@ bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t stateCountFunction(SqlFunctionCtx* pCtx);
 int32_t stateDurationFunction(SqlFunctionCtx* pCtx);
 
+bool getCsumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+int32_t csumFunction(SqlFunctionCtx* pCtx);
+
 bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 
 #ifdef __cplusplus

source/libs/function/src/builtins.c

@@ -308,6 +308,37 @@ static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateCsum(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  if (1 != LIST_LENGTH(pFunc->pParameterList)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
+  if (QUERY_NODE_COLUMN != nodeType(pPara)) {
+    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
+                           "The input parameter of CSUM function can only be column");
+  }
+
+  uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
+  uint8_t resType;
+  if (!IS_NUMERIC_TYPE(colType)) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  } else {
+    if (IS_SIGNED_NUMERIC_TYPE(colType)) {
+      resType = TSDB_DATA_TYPE_BIGINT;
+    } else if (IS_UNSIGNED_NUMERIC_TYPE(colType)) {
+      resType = TSDB_DATA_TYPE_UBIGINT;
+    } else if (IS_FLOAT_TYPE(colType)) {
+      resType = TSDB_DATA_TYPE_DOUBLE;
+    } else {
+      ASSERT(0);
+    }
+  }
+
+  pFunc->node.resType = (SDataType) { .bytes = tDataTypes[resType].bytes, .type = resType};
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateLastRow(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // todo
   return TSDB_CODE_SUCCESS;
@@ -742,6 +773,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = stateDurationFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "csum",
+    .type = FUNCTION_TYPE_CSUM,
+    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .translateFunc = translateCsum,
+    .getEnvFunc   = getCsumFuncEnv,
+    .initFunc     = functionSetup,
+    .processFunc  = csumFunction,
+    .finalizeFunc = NULL
+  },
   {
     .name = "abs",
     .type = FUNCTION_TYPE_ABS,

source/libs/function/src/builtinsimpl.c

@@ -2818,7 +2818,6 @@ int32_t stateCountFunction(SqlFunctionCtx* pCtx) {
   SInputColumnInfoData* pInput = &pCtx->input;
 
   SColumnInfoData* pInputCol = pInput->pData[0];
-  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
 
   int32_t numOfElems = 0;
   SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
@@ -2856,7 +2855,6 @@ int32_t stateDurationFunction(SqlFunctionCtx* pCtx) {
   TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
 
   SColumnInfoData* pInputCol = pInput->pData[0];
-  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
 
   int32_t numOfElems = 0;
   SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
@@ -2896,3 +2894,58 @@ int32_t stateDurationFunction(SqlFunctionCtx* pCtx) {
 
   return numOfElems;
 }
+
+bool getCsumFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SSumRes);
+  return true;
+}
+
+int32_t csumFunction(SqlFunctionCtx* pCtx) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SSumRes*             pSumRes = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
+
+  SColumnInfoData* pInputCol = pInput->pData[0];
+  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
+  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
+
+  int32_t numOfElems = 0;
+  int32_t type = pInputCol->info.type;
+  int32_t startOffset = pCtx->offset;
+  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
+    int32_t pos = startOffset + numOfElems;
+    if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+      //colDataAppendNULL(pOutput, i);
+      continue;
+    }
+
+    char* data = colDataGetData(pInputCol, i);
+    if (IS_SIGNED_NUMERIC_TYPE(type)) {
+      int64_t v;
+      GET_TYPED_DATA(v, int64_t, type, data);
+      pSumRes->isum += v;
+      colDataAppend(pOutput, pos, (char *)&pSumRes->isum, false);
+    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
+      uint64_t v;
+      GET_TYPED_DATA(v, uint64_t, type, data);
+      pSumRes->usum += v;
+      colDataAppend(pOutput, pos, (char *)&pSumRes->usum, false);
+    } else if (IS_FLOAT_TYPE(type)) {
+      double v;
+      GET_TYPED_DATA(v, double, type, data);
+      pSumRes->dsum += v;
+      colDataAppend(pOutput, pos, (char *)&pSumRes->dsum, false);
+    }
+
+    //TODO: remove this after pTsOutput is handled
+    if (pTsOutput != NULL) {
+      colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+    }
+
+    numOfElems++;
+  }
+
+  return numOfElems;
+}

tests/pytest/util/sql.py

@@ -150,6 +150,9 @@ class TDSql:
             raise Exception(repr(e))
         return (self.queryRows, timeout)
 
+    def getRows(self):
+        return self.queryRows
+
     def checkRows(self, expectRows):
         if self.queryRows == expectRows:
             tdLog.info("sql:%s, queryRows:%d == expect:%d" % (self.sql, self.queryRows, expectRows))

tests/script/tsim/stable/alter1.sim

@@ -159,6 +159,7 @@ sql alter table db.stb rename tag t1 tx
 
 print ========== alter common
 sql alter table db.stb comment 'abcde' ;
+sql alter table db.stb ttl 10 ;
 
 sql show db.stables;
 if $data[0][6] != abcde then

tests/system-test/1-insert/insertWithMoreVgroup.py

@@ -346,8 +346,10 @@ class TDTestCase:
         return
 
     def test_case3(self):
+        self.taosBenchCreate("127.0.0.1","no","db1", "stb1", 1, 8, 1*10000)
+        # self.taosBenchCreate("test209","no","db2", "stb2", 1, 8, 1*10000)
+
         # self.taosBenchCreate("chenhaoran02","no","db1", "stb1", 1, 8, 1*10000)
-        self.taosBenchCreate("chenhaoran02","no","db1", "stb1", 1, 8, 1*1000)
 
         # self.taosBenchCreate("db1", "stb1", 4, 5, 100*10000)
         # self.taosBenchCreate("db1", "stb1", 1, 5, 100*10000)

tests/system-test/1-insert/manyVgroups.json

@@ -29,8 +29,8 @@
                     "batch_create_tbl_num": 50000,
                     "data_source": "rand",
                     "insert_mode": "taosc",
-                    "insert_rows": 0,
+                    "insert_rows": 10,
-                    "interlace_rows": 0,
+                    "interlace_rows": 100000,
                     "insert_interval": 0,
                     "max_sql_len": 10000000,
                     "disorder_ratio": 0,