Merge remote-tracking branch 'origin/3.0' into fix/TD-19963

Jenkinsfile2

@@ -173,8 +173,8 @@ def pre_test_build_mac() {
     '''
     sh '''
         cd ${WK}/debug
-        cmake .. -DBUILD_TEST=true
-        make -j8
+        cmake .. -DBUILD_TEST=true -DBUILD_HTTPS=false
+        make -j10
     '''
     sh '''
         date
@@ -428,7 +428,7 @@ pipeline {
                                         date
                                         rm -rf ${WKC}/debug
                                         cd ${WKC}/tests/parallel_test
-                                        time ./container_build.sh -w ${WKDIR} -t 8 -e
+                                        time ./container_build.sh -w ${WKDIR} -t 10 -e
                                         rm -f /tmp/cases.task
                                         ./collect_cases.sh -e
                                     '''

include/libs/stream/streamState.h

@@ -99,6 +99,8 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
 
+// char* streamStateSessionDump(SStreamState* pState);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -124,6 +124,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_VGROUP_LIST, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_REDISTRIBUTE_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_MERGE_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_SPLIT_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_BALANCE_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_FUNC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_RETRIEVE_FUNC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@@ -60,7 +60,7 @@ int32_t mndInitVgroup(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_RSP, mndTransProcessRsp);
 
   mndSetMsgHandle(pMnode, TDMT_MND_REDISTRIBUTE_VGROUP, mndProcessRedistributeVgroupMsg);
-  mndSetMsgHandle(pMnode, TDMT_MND_MERGE_VGROUP, mndProcessSplitVgroupMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_SPLIT_VGROUP, mndProcessSplitVgroupMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_BALANCE_VGROUP, mndProcessBalanceVgroupMsg);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndRetrieveVgroups);
@@ -617,35 +617,6 @@ SEpSet mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup) {
   return epset;
 }
 
-static bool mndGetVgroupMaxReplicaFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
-  SVgObj  *pVgroup = pObj;
-  int64_t  uid = *(int64_t *)p1;
-  int8_t  *pReplica = p2;
-  int32_t *pNumOfVgroups = p3;
-
-  if (pVgroup->dbUid == uid) {
-    *pReplica = TMAX(*pReplica, pVgroup->replica);
-    (*pNumOfVgroups)++;
-  }
-
-  return true;
-}
-
-static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pReplica, int32_t *pNumOfVgroups) {
-  SSdb   *pSdb = pMnode->pSdb;
-  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
-  if (pDb == NULL) {
-    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
-    return -1;
-  }
-
-  *pReplica = 1;
-  *pNumOfVgroups = 0;
-  sdbTraverse(pSdb, SDB_VGROUP, mndGetVgroupMaxReplicaFp, &pDb->uid, pReplica, pNumOfVgroups);
-  mndReleaseDb(pMnode, pDb);
-  return 0;
-}
-
 static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
   SMnode *pMnode = pReq->info.node;
   SSdb   *pSdb = pMnode->pSdb;

source/dnode/mnode/impl/test/sdb/sdbTest.cpp

@@ -367,6 +367,80 @@ bool sdbTraverseFail(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, in
   return false;
 }
 
+#ifndef WINDOWS
+
+TEST_F(MndTestSdb, 00_API) {
+  SMnode  mnode = {0};
+  SSdbOpt opt = {0};
+  opt.pMnode = &mnode;
+  opt.path = TD_TMP_DIR_PATH "mnode_test_sdb";
+  taosRemoveDir(opt.path);
+  SSdb *pSdb = sdbInit(&opt);
+
+  SSdbTable table = {.sdbType = SDB_USER, .keyType = SDB_KEY_BINARY};
+  sdbSetTable(pSdb, table);
+
+  // sdbRow.c
+  SSdbRow *pRow1 = sdbAllocRow(-128);
+  ASSERT_EQ(pRow1 == NULL, 1);
+
+  void *pRow2 = sdbGetRowObj(NULL);
+  ASSERT_EQ(pRow2 == NULL, 1);
+
+  //sdbRaw.c
+  SStrObj  strObj;
+  SSdbRaw *pRaw1 = NULL;
+  strSetDefault(&strObj, 1);
+  
+  pRaw1 = strEncode(&strObj);
+  int32_t id = sdbGetIdFromRaw(pSdb, pRaw1);
+  ASSERT_EQ(id, -2);
+
+  SSdbRaw *pRaw2 = sdbAllocRaw(SDB_USER, 1, -128);
+  ASSERT_EQ(pRaw2 == NULL, 1);
+
+  ASSERT_EQ(sdbSetRawInt8(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbSetRawInt8(pRaw1, -128, 0), -1);
+  ASSERT_EQ(sdbSetRawInt32(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbSetRawInt32(pRaw1, -128, 0), -1);
+  ASSERT_EQ(sdbSetRawInt16(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbSetRawInt16(pRaw1, -128, 0), -1);
+  ASSERT_EQ(sdbSetRawInt64(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbSetRawInt64(pRaw1, -128, 0), -1);
+  ASSERT_EQ(sdbSetRawBinary(NULL, 0, "12", 3), -1);
+  ASSERT_EQ(sdbSetRawBinary(pRaw1, 9028, "12", 3), -1);
+  ASSERT_EQ(sdbSetRawDataLen(NULL, 0), -1);
+  ASSERT_EQ(sdbSetRawDataLen(pRaw1, 9000), -1);
+  ASSERT_EQ(sdbSetRawStatus(NULL, SDB_STATUS_READY), -1);
+  ASSERT_EQ(sdbSetRawStatus(pRaw1, SDB_STATUS_INIT), -1);
+
+  ASSERT_EQ(sdbGetRawInt8(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbGetRawInt8(pRaw1, 9000, 0), -1);
+  ASSERT_EQ(sdbGetRawInt32(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbGetRawInt32(pRaw1, 9000, 0), -1);
+  ASSERT_EQ(sdbGetRawInt16(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbGetRawInt16(pRaw1, 9000, 0), -1);
+  ASSERT_EQ(sdbGetRawInt64(NULL, 0, 0), -1);
+  ASSERT_EQ(sdbGetRawInt64(pRaw1, 9000, 0), -1);
+  ASSERT_EQ(sdbGetRawBinary(NULL, 0, 0, 4096), -1);
+  ASSERT_EQ(sdbGetRawBinary(pRaw1, 9000, 0, 112), -1);
+  ASSERT_EQ(sdbGetRawSoftVer(NULL, 0), -1);
+  ASSERT_EQ(sdbGetRawTotalSize(NULL), -1);
+
+  // sdbHash.c
+  EXPECT_STREQ(sdbTableName((ESdbType)100), "undefine");
+  EXPECT_STREQ(sdbStatusName((ESdbStatus)100), "undefine");
+  ASSERT_EQ(sdbGetTableVer(pSdb, (ESdbType)100), -1);
+
+  SSdbRaw *pRaw3 = sdbAllocRaw((ESdbType)-12, 1, 128);
+  ASSERT_NE(sdbWriteWithoutFree(pSdb, pRaw3), 0);
+  pSdb->hashObjs[1] = NULL;
+  SSdbRaw *pRaw4 = sdbAllocRaw((ESdbType)1, 1, 128);
+  ASSERT_NE(sdbWriteWithoutFree(pSdb, pRaw4), 0);
+}
+
+#endif
+
 TEST_F(MndTestSdb, 01_Write_Str) {
   void    *pIter = NULL;
   int32_t  num = 0;
@@ -885,8 +959,8 @@ TEST_F(MndTestSdb, 01_Read_Str) {
     ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
     pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
     ASSERT_EQ(pI32Obj, nullptr);
-    int32_t code = terrno;
-    ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_DROPPING);
+    // int32_t code = terrno;
+    // ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_DROPPING);
   }
 
   {
@@ -899,8 +973,8 @@ TEST_F(MndTestSdb, 01_Read_Str) {
     ASSERT_EQ(sdbWrite(pSdb, pRaw), 0);
     pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key);
     ASSERT_EQ(pI32Obj, nullptr);
-    int32_t code = terrno;
-    ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_CREATING);
+    // int32_t code = terrno;
+    // ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_CREATING);
   }
 
   {
@@ -934,4 +1008,4 @@ TEST_F(MndTestSdb, 01_Read_Str) {
   sdbCleanup(pSdb);
   ASSERT_EQ(mnode.insertTimes, 9);
   ASSERT_EQ(mnode.deleteTimes, 9);
-}
\ No newline at end of file
+}

source/dnode/mnode/sdb/src/sdbFile.c

@@ -654,15 +654,12 @@ int32_t sdbStopWrite(SSdb *pSdb, SSdbIter *pIter, bool isApply, int64_t index, i
   }
 
   if (config > 0) {
-    ASSERT(pSdb->commitConfig == config);
     pSdb->commitConfig = config;
   }
   if (term > 0) {
-    ASSERT(pSdb->commitTerm == term);
     pSdb->commitTerm = term;
   }
   if (index > 0) {
-    ASSERT(pSdb->commitIndex == index);
     pSdb->commitIndex = index;
   }
 

source/dnode/mnode/sdb/src/sdbHash.c

@@ -241,9 +241,7 @@ int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw) {
 
   SdbDecodeFp decodeFp = pSdb->decodeFps[pRaw->type];
   SSdbRow    *pRow = (*decodeFp)(pRaw);
-  if (pRow == NULL) {
-    return terrno;
-  }
+  if (pRow == NULL) return terrno;
 
   pRow->type = pRaw->type;
 

source/dnode/vnode/src/tq/tqSink.c

@@ -42,7 +42,7 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
     if (metaGetTableEntryByName(&mr, name) < 0) {
       metaReaderClear(&mr);
       taosMemoryFree(name);
-      return -1;
+      continue;
     }
 
     int64_t uid = mr.me.uid;

source/libs/executor/src/scanoperator.c

@@ -1299,6 +1299,9 @@ static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSr
     SSessionKey startWin = {0};
     getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], endData[i], groupId, &startWin);
     if (IS_INVALID_SESSION_WIN_KEY(startWin)) {
+      // char* tmp = streamStateSessionDump(pInfo->windowSup.pStreamAggSup->pState);
+      // qInfo("%s", tmp);
+      // taosMemoryFree(tmp);
       // window has been closed.
       continue;
     }

source/libs/function/src/builtinsimpl.c

@@ -380,7 +380,7 @@ typedef struct SGroupKeyInfo {
       numOfElem += 1;                                              \
       pStddevRes->count -= 1;                                      \
       sumT -= plist[i];                                            \
-      pStddevRes->quadraticISum -= plist[i] * plist[i];            \
+      pStddevRes->quadraticISum -= (int64_t)(plist[i] * plist[i]); \
     }                                                              \
   } while (0)
 
@@ -2526,8 +2526,9 @@ int32_t apercentileFunction(SqlFunctionCtx* pCtx) {
     // might be a race condition here that pHisto can be overwritten or setup function
     // has not been called, need to relink the buffer pHisto points to.
     buildHistogramInfo(pInfo);
-    qDebug("%s before add %d elements into histogram, total:%" PRId64 ", numOfEntry:%d, pHisto:%p, elems: %p", __FUNCTION__,
-           numOfElems, pInfo->pHisto->numOfElems, pInfo->pHisto->numOfEntries, pInfo->pHisto, pInfo->pHisto->elems);
+    qDebug("%s before add %d elements into histogram, total:%" PRId64 ", numOfEntry:%d, pHisto:%p, elems: %p",
+           __FUNCTION__, numOfElems, pInfo->pHisto->numOfElems, pInfo->pHisto->numOfEntries, pInfo->pHisto,
+           pInfo->pHisto->elems);
     for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
       if (colDataIsNull_f(pCol->nullbitmap, i)) {
         continue;

source/libs/stream/src/streamState.c

@@ -497,7 +497,7 @@ SStreamStateCur* streamStateSessionGetRanomCur(SStreamState* pState, const SSess
   if (pCur == NULL) return NULL;
   tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL);
 
-  int32_t          c = 0;
+  int32_t          c = -2;
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
   tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c);
   if (c != 0) {
@@ -727,3 +727,43 @@ _end:
   streamStateFreeCur(pCur);
   return res;
 }
+
+#if 0
+char* streamStateSessionDump(SStreamState* pState) {
+  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
+  if (pCur == NULL) {
+    return NULL;
+  }
+  pCur->number = pState->number;
+  if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+  tdbTbcMoveToFirst(pCur->pCur);
+
+  SSessionKey key = {0};
+  int32_t     code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
+  if (code != 0) {
+    return NULL;
+  }
+
+  int32_t size = 2048;
+  char*   dumpBuf = taosMemoryCalloc(size, 1);
+  int64_t len = 0;
+  len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.win.skey);
+  len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
+  len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
+  while (1) {
+    tdbTbcMoveToNext(pCur->pCur);
+    key = (SSessionKey){0};
+    code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
+    if (code != 0) {
+      return dumpBuf;
+    }
+    len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.win.skey);
+    len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
+    len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
+  }
+  return dumpBuf;
+}
+#endif

tests/parallel_test/container_build.sh

@@ -52,7 +52,7 @@ fi
 
 docker run \
     -v $REP_MOUNT_PARAM \
-    --rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DWEBSOCKET=true;make -j $THREAD_COUNT"
+    --rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true;make -j $THREAD_COUNT"
 
 ret=$?
 exit $ret

tests/script/jenkins/basic.txt

@@ -227,7 +227,7 @@
 ./test.sh -f tsim/table/vgroup.sim
 
 # ---- stream
-./test.sh -f tsim/stream/basic0.sim -v
+./test.sh -f tsim/stream/basic0.sim -g
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
 ./test.sh -f tsim/stream/drop_stream.sim
@@ -446,5 +446,5 @@
 ./test.sh -f tsim/tag/tinyint.sim
 ./test.sh -f tsim/tag/drop_tag.sim
 
-
+./test.sh -f tmp/monitor.sim
 #======================b1-end===============

tests/script/test.sh

@@ -15,7 +15,7 @@ VALGRIND=0
 UNIQUE=0
 UNAME_BIN=`which uname`
 OS_TYPE=`$UNAME_BIN`
-while getopts "f:avum" arg
+while getopts "f:agvum" arg
 do
   case $arg in
     f)
@@ -27,6 +27,9 @@ do
     u)
       UNIQUE=1
       ;;
+    g)
+      VALGRIND=2
+      ;;  
     ?)
       echo "unknow argument"
       ;;
@@ -125,6 +128,9 @@ if [ -n "$FILE_NAME" ]; then
     FLAG="-v"
     echo valgrind --tool=memcheck --leak-check=full --show-reachable=no  --track-origins=yes --child-silent-after-fork=yes --show-leak-kinds=all --num-callers=20 -v  --workaround-gcc296-bugs=yes  --log-file=${LOG_DIR}/valgrind-tsim.log $PROGRAM -c $CFG_DIR -f $FILE_NAME $FLAG
     valgrind --tool=memcheck --leak-check=full --show-reachable=no  --track-origins=yes --child-silent-after-fork=yes --show-leak-kinds=all --num-callers=20 -v  --workaround-gcc296-bugs=yes  --log-file=${LOG_DIR}/valgrind-tsim.log $PROGRAM -c $CFG_DIR -f $FILE_NAME $FLAG
+  elif [ $VALGRIND -eq 2 ]; then
+    echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME -v
+    $PROGRAM -c $CFG_DIR -f $FILE_NAME -v
   else
     echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME  
     $PROGRAM -c $CFG_DIR -f $FILE_NAME

tests/script/tmp/monitor.sim

@@ -24,3 +24,5 @@ print =============== create drop qnode 1
 sql create qnode on dnode 1
 #sql create snode on dnode 1
 
+sleep 10000
+system sh/exec.sh -n dnode1 -s stop

tests/script/tsim/insert/basic.sim

@@ -50,6 +50,7 @@ print $data[1000][1000]
 $a1 = 0
 $a2 = 0
 $a3 = $a1 % $a2
+$a4 = $a1 / $a2
 print $a3 
 
 $val = \\\1

tests/script/tsim/stream/partitionbyColumnInterval.sim

@@ -599,12 +599,62 @@ if $loop_count == 20 then
 endi
 
 if $rows != 4 then
-  print =====rows=$rows
-  print =====rows=$rows
-  print =====rows=$rows
-  #goto loop15
+  print ======rows=$rows
+  goto loop15
+endi
+
+sql insert into t1 values(1648791223001,11,2,3,1.0);
+sql insert into t2 values(1648791223001,21,2,3,1.0);
+sql insert into t3 values(1648791223001,31,2,3,1.0);
+sql insert into t4 values(1648791223001,41,2,3,1.0);
+
+sleep 300
+
+sql delete from st where ts = 1648791223001;
+
+$loop_count = 0
+
+loop16:
+sleep 50
+sql select * from test.streamt5 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  print ======rows=$rows
+  return -1
+endi
+
+if $rows != 4 then
+  goto loop16
+endi
+
+sql insert into t1 values(1648791223001,12,2,3,1.0);
+sql insert into t2 values(1648791223001,22,2,3,1.0);
+sql insert into t3 values(1648791223001,32,2,3,1.0);
+sql insert into t4 values(1648791223001,42,2,3,1.0);
+
+sleep 300
+
+sql delete from st where ts = 1648791223001;
+
+$loop_count = 0
+
+loop17:
+sleep 50
+sql select * from test.streamt5 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
 endi
 
+if $rows != 4 then
+  print ======rows=$rows
+  print ======rows=$rows
+  print ======rows=$rows
+  return 1
+  #goto loop17
+endi
 
 $loop_all = $loop_all + 1
 print ============loop_all=$loop_all