Merge pull request #6487 from taosdata/feature/m2d

Feature/m2d

src/balance/src/bnMain.c

@@ -220,10 +220,6 @@ int32_t bnAllocVnodes(SVgObj *pVgroup) {
 }
 
 static bool bnCheckVgroupReady(SVgObj *pVgroup, SVnodeGid *pRmVnode) {
-  if (pVgroup->lbTime + 5 * tsStatusInterval > tsAccessSquence) {
-    return false;
-  }
-
   int32_t rmVnodeVer = 0;
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
     SVnodeGid *pVnode = pVgroup->vnodeGid + i;
@@ -405,7 +401,7 @@ void bnReset() {
     if (pDnode == NULL) break;
 
     // while master change, should reset dnode to offline
-    mInfo("dnode:%d set access:%d to 0", pDnode->dnodeId, pDnode->lastAccess);
+    mInfo("dnode:%d set access:%" PRId64 " to 0", pDnode->dnodeId, pDnode->lastAccess);
     pDnode->lastAccess = 0;
     if (pDnode->status != TAOS_DN_STATUS_DROPPING) {
       pDnode->status = TAOS_DN_STATUS_OFFLINE;
@@ -499,7 +495,7 @@ static bool bnMontiorDropping() {
       if (dnodeIsMasterEp(pDnode->dnodeEp)) continue; 
       if (mnodeGetDnodesNum() <= 1) continue;
 
-      mLInfo("dnode:%d, set to removing state for it offline:%d seconds", pDnode->dnodeId,
+      mLInfo("dnode:%d, set to removing state for it offline:%" PRId64 " seconds", pDnode->dnodeId,
               tsAccessSquence - pDnode->lastAccess);
 
       pDnode->status = TAOS_DN_STATUS_DROPPING;
@@ -574,8 +570,8 @@ void bnCheckStatus() {
       if (pDnode->status != TAOS_DN_STATUS_DROPPING && pDnode->status != TAOS_DN_STATUS_OFFLINE) {
         pDnode->status = TAOS_DN_STATUS_OFFLINE;
         pDnode->offlineReason = TAOS_DN_OFF_STATUS_MSG_TIMEOUT;
-        mInfo("dnode:%d, set to offline state, access seq:%d last seq:%d laststat:%d", pDnode->dnodeId, tsAccessSquence,
-              pDnode->lastAccess, pDnode->status);
+        mInfo("dnode:%d, set to offline state, access seq:%" PRId64 " last seq:%" PRId64 " laststat:%d", pDnode->dnodeId,
+              tsAccessSquence, pDnode->lastAccess, pDnode->status);
         bnSetVgroupOffline(pDnode);
         bnStartTimer(3000);
       }

src/balance/src/bnThread.c

@@ -102,12 +102,12 @@ static void bnProcessTimer(void *handle, void *tmrId) {
   if (tsBnThread.stop) return;
 
   tsBnThread.timer = NULL;
-  tsAccessSquence++;
-
   bnStartTimer(-1);
   bnCheckStatus();
 
   if (handle == NULL) {
+    ++tsAccessSquence;
+
     if (tsAccessSquence % tsBalanceInterval == 0) {
       mDebug("balance function is scheduled by timer");
       bnPostSignal();
@@ -122,8 +122,7 @@ static void bnProcessTimer(void *handle, void *tmrId) {
 void bnStartTimer(int32_t mseconds) {
   if (tsBnThread.stop) return;
 
-  bool updateSoon = (mseconds != -1);
-  if (updateSoon) {
+  if (mseconds != -1) {
     mTrace("balance function will be called after %d ms", mseconds);
     taosTmrReset(bnProcessTimer, mseconds, (void *)(int64_t)mseconds, tsMnodeTmr, &tsBnThread.timer);
   } else {
@@ -132,5 +131,5 @@ void bnStartTimer(int32_t mseconds) {
 }
 
 void bnNotify() {
-  bnStartTimer(500); 
+  bnStartTimer(500);
 }

src/mnode/inc/mnodeDef.h

@@ -48,9 +48,8 @@ typedef struct SDnodeObj {
   int32_t    dnodeId;
   int32_t    openVnodes;
   int64_t    createdTime;
-  int32_t    resever0;         // from dnode status msg, config information
+  int64_t    lastAccess;
   int32_t    customScore;      // config by user
-  uint32_t   lastAccess;
   uint16_t   numOfCores;       // from dnode status msg
   uint16_t   dnodePort;
   char       dnodeFqdn[TSDB_FQDN_LEN];

src/mnode/inc/mnodeDnode.h

@@ -78,7 +78,7 @@ void    mnodeUpdateDnode(SDnodeObj *pDnode);
 int32_t mnodeDropDnode(SDnodeObj *pDnode, void *pMsg);
 
 int32_t mnodeCompactDnodes();
-extern int32_t tsAccessSquence;
+extern int64_t tsAccessSquence;
 
 #ifdef __cplusplus
 }

src/mnode/src/mnodeDnode.c

@@ -39,8 +39,8 @@
 #include "mnodePeer.h"
 #include "mnodeCluster.h"
 
-int32_t tsAccessSquence = 0;
-int64_t         tsDnodeRid = -1;
+int64_t        tsAccessSquence = 0;
+int64_t        tsDnodeRid = -1;
 static void *  tsDnodeSdb = NULL;
 static int32_t tsDnodeUpdateSize = 0;
 extern void *  tsMnodeSdb;
@@ -567,7 +567,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
              mnodeGetClusterId());
       return TSDB_CODE_MND_INVALID_CLUSTER_ID;
     } else {
-      mTrace("dnode:%d, status received, access times %d openVnodes:%d:%d", pDnode->dnodeId, pDnode->lastAccess,
+      mTrace("dnode:%d, status received, access times %" PRId64 " openVnodes:%d:%d", pDnode->dnodeId, pDnode->lastAccess,
              htons(pStatus->openVnodes), pDnode->openVnodes);
     }
   }
@@ -629,9 +629,9 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
     bnNotify();
   }
 
-  if (!tsEnableBalance) {
-    int32_t numOfMnodes = mnodeGetMnodesNum();
-    if (numOfMnodes < tsNumOfMnodes) bnNotify();
+  int32_t numOfMnodes = mnodeGetMnodesNum();
+  if (numOfMnodes < tsNumOfMnodes && numOfMnodes < mnodeGetOnlineDnodesNum() && !pDnode->isMgmt) {
+    bnNotify();
   }
 
   if (openVnodes != pDnode->openVnodes) {

src/mnode/src/mnodeSdb.c

@@ -670,10 +670,17 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void *
                pTable->name, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
       return TSDB_CODE_SUCCESS;
     } else if (pHead->version != tsSdbMgmt.version + 1) {
-      pthread_mutex_unlock(&tsSdbMgmt.mutex);
-      sdbError("vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64,
-               pTable->name, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
-      return TSDB_CODE_SYN_INVALID_VERSION;
+      if (qtype != TAOS_QTYPE_WAL) {
+        pthread_mutex_unlock(&tsSdbMgmt.mutex);
+        sdbError(
+            "vgId:1, sdb:%s, failed to restore %s key:%s from source(%d), hver:%" PRIu64 " too large, mver:%" PRIu64,
+            pTable->name, actStr[action], sdbGetKeyStr(pTable, pHead->cont), qtype, pHead->version, tsSdbMgmt.version);
+        return TSDB_CODE_SYN_INVALID_VERSION;
+      } else {
+        // If cksum is wrong when recovering wal, use this code
+        tsSdbMgmt.version = pHead->version;
+      }
+
     } else {
       tsSdbMgmt.version = pHead->version;
     }

src/wal/src/walWrite.c

@@ -346,7 +346,7 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
     }
 
 #if defined(WAL_CHECKSUM_WHOLE)
-    if (pHead->sver == 0 && !walValidateChecksum(pHead)) {
+    if ((pHead->sver == 0 && !walValidateChecksum(pHead)) || pHead->sver < 0 || pHead->sver > 1) {
       wError("vgId:%d, file:%s, wal head cksum is messed up, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
              pHead->version, pHead->len, offset);
       code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);

tests/pytest/fulltest.sh

@@ -21,6 +21,7 @@ python3 insert/retentionpolicy.py
 python3 ./test.py -f insert/alterTableAndInsert.py
 python3 ./test.py -f insert/insertIntoTwoTables.py
 python3 ./test.py -f insert/before_1970.py
+python3 ./test.py -f insert/special_character_show.py
 python3 bug2265.py
 python3 ./test.py -f insert/bug3654.py
 python3 ./test.py -f insert/insertDynamicColBeforeVal.py

tests/pytest/insert/restfulInsert.py

@@ -40,12 +40,13 @@ class RestfulInsert:
             if tableID + i >= self.numOfTables : break
             name = 'beijing' if (tableID + i) % 2 == 0 else 'shanghai'
             data = "create table if not exists %s.%s%d using %s.meters tags(%d, '%s')" % (self.dbname, self.tableNamePerfix, tableID + i, self.dbname, tableID + i, name)
-            response = requests.post(self.url, data, headers = self.header)
-            if response.status_code != 200:
-                    print(response.content)
-
+            try:
+                response = requests.post(self.url, data, headers = self.header)
+                if response.status_code != 200:
+                    print(response.content)                
+            except Exception as e:
+                print(e)
 
-   
     def insertData(self, threadID):        
         print("thread %d started" % threadID)
         tablesPerThread = int (self.numOfTables / self.numOfThreads)   
@@ -90,10 +91,16 @@ class RestfulInsert:
                     if len(data) > 1024*1024 : 
                         print ('batch size is larger than 1M')
                         exit(-1)
-                    response = requests.post(self.url, data, headers = self.header)
-                    if response.status_code != 200:
-                        print(response.content)
-
+                    try:
+                        startTime = time.time()
+                        response = requests.post(self.url, data, headers = self.header)
+                        endTime = time.time()
+                        if response.status_code != 200:
+                            print(response.content)
+                        else:
+                            print("inserted %d records, %d seconds" % (bloop, endTime - startTime))
+                    except Exception as e:
+                        print(e)
 
     def insertUnlimitedData(self, threadID):        
         print("thread %d started" % threadID)
@@ -119,10 +126,17 @@ class RestfulInsert:
                 else:
                     random.shuffle(values)
                     for k in range(len(values)):            
-                        data += values[k]                
-                response = requests.post(self.url, data, headers = self.header)
-                if response.status_code != 200:
-                    print(response.content)
+                        data += values[k]    
+                try:     
+                    startTime = time.time()
+                    response = requests.post(self.url, data, headers = self.header)
+                    endTime = time.time()
+                    if response.status_code != 200:
+                        print(response.content)
+                    else:
+                        print("inserted %d records, %d seconds" % (self.batchSize, endTime - startTime))
+                except Exception as e:
+                    print(e)                    
 
     def run(self):            
         data = "create database if not exists %s" % self.dbname

tests/pytest/insert/special_character_show.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+from util.log import *
+from util.cases import *
+from util.sql import *
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+
+    def run(self):
+        tdSql.prepare()
+        # test case for https://jira.taosdata.com:18080/browse/TD-4584
+
+        #1
+        tdLog.info('=============== step1,create stable')
+        tdLog.info('create table stb1 (ts timestamp, value double) tags (bin binary(128))') 
+        tdSql.execute('create table stb1 (ts timestamp, value double) tags (bin binary(128))')
+
+        tdLog.info('=============== step2,create table增加了转义字符')
+        tdLog.info('create table tb1 using stb1 tags("abc\\"def")')
+        #增加了转义字符\
+        tdSql.execute('create table tb1 using stb1 tags("abc\\"def")')
+
+        tdLog.info('=============== step3,insert data') 
+        tdLog.info('insert into  tb1 values(now,1.0)')
+        tdSql.execute('insert into  tb1 values(now,1.0)')
+
+        tdLog.info('=============== step4,select table') 
+        tdLog.info('select * from stb1 ')
+        tdSql.query('select * from stb1 ')
+
+        tdLog.info('=============== step5,check data') 
+        tdSql.checkData(0,2,'abc"def')
+
+        
+  
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())