Merge pull request #860 from taosdata/hotfix/TBASE-1284

[TBASE-1284]

src/sdb/src/sdbEngine.c

@@ -171,6 +171,7 @@ int sdbInitTableByFile(SSdbTable *pTable) {
   void *   pMetaRow = NULL;
   int      total_size = 0;
   int      real_size = 0;
+  int      maxAutoIndex = 0;
 
   oldId = pTable->id;
   if (sdbOpenSdbFile(pTable) < 0) return -1;
@@ -240,10 +241,18 @@ int sdbInitTableByFile(SSdbTable *pTable) {
         rowMeta.rowSize = rowHead->rowSize;
         rowMeta.row = (*(pTable->appTool))(SDB_TYPE_DECODE, NULL, rowHead->data, rowHead->rowSize, NULL);
         (*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, rowMeta.row, &rowMeta);
-        if (pTable->keyType == SDB_KEYTYPE_AUTO) pTable->autoIndex++;
+        if (pTable->keyType == SDB_KEYTYPE_AUTO) {
+          pTable->autoIndex++;
+          maxAutoIndex = MAX(maxAutoIndex, *(int32_t*)rowHead->data);
+        }
         pTable->numOfRows++;
       }
     } else {                  // already exists
+      if (pTable->keyType == SDB_KEYTYPE_AUTO) {
+        pTable->autoIndex++;
+        maxAutoIndex = MAX(maxAutoIndex, *(int32_t *) rowHead->data);
+      }
+
       if (rowHead->id < 0) {  // Delete the object
         (*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, rowHead->data);
         (*(pTable->appTool))(SDB_TYPE_DESTROY, pMetaRow, NULL, 0, NULL);
@@ -260,6 +269,10 @@ int sdbInitTableByFile(SSdbTable *pTable) {
     if (pTable->id < abs(rowHead->id)) pTable->id = abs(rowHead->id);
   }
 
+  if (pTable->keyType == SDB_KEYTYPE_AUTO) {
+    pTable->autoIndex = maxAutoIndex;
+  }
+
   sdbVersion += (pTable->id - oldId);
   if (numOfDels > pTable->maxRows / 4) sdbSaveSnapShot(pTable);
 
@@ -372,10 +385,10 @@ int64_t sdbInsertRow(void *handle, void *row, int rowSize) {
             sdbError("table:%s, failed to insert record:%s sdbVersion:%ld id:%d", pTable->name, taosIpStr(*(int32_t *)row), sdbVersion, pTable->id);
             break;
           case SDB_KEYTYPE_AUTO:
-            sdbError("table:%s, failed to insert record:%s sdbVersion:%ld id:%d", pTable->name, *(int32_t *)row, sdbVersion, pTable->id);
+            sdbError("table:%s, failed to insert record:%d sdbVersion:%ld id:%d", pTable->name, *(int32_t *)row, sdbVersion, pTable->id);
             break;
           default:
-            sdbError("table:%s, failed to insert record:%s sdbVersion:%ld id:%d", pTable->name, sdbVersion, pTable->id);
+            sdbError("table:%s, failed to insert record sdbVersion:%ld id:%d", pTable->name, sdbVersion, pTable->id);
             break;
         }
         return -1;
@@ -593,15 +606,15 @@ int sdbUpdateRow(void *handle, void *row, int updateSize, char isUpdated) {
                 pTable->name, (char *) row, sdbVersion, pTable->id);
         break;
       case SDB_KEYTYPE_UINT32: //dnodes or mnodes
-        sdbError("table:%s, failed to update record:%s record is not there, sdbVersion:%ld id:%d",
+        sdbError("table:%s, failed to update record:%s, record is not there, sdbVersion:%ld id:%d",
                 pTable->name, taosIpStr(*(int32_t *) row), sdbVersion, pTable->id);
         break;
       case SDB_KEYTYPE_AUTO:
-        sdbError("table:%s, failed to update record:F%s record is not there, sdbVersion:%ld id:%d",
+        sdbError("table:%s, failed to update record:%d, record is not there, sdbVersion:%ld id:%d",
                 pTable->name, *(int32_t *) row, sdbVersion, pTable->id);
         break;
       default:
-        sdbError("table:%s, failed to update record:%s record is not there, sdbVersion:%ld id:%d",
+        sdbError("table:%s, failed to update record, record is not there, sdbVersion:%ld id:%d",
                 pTable->name, sdbVersion, pTable->id);
         break;
     }

src/system/detail/src/mgmtMeter.c

@@ -233,6 +233,10 @@ void *mgmtMeterActionDelete(void *row, char *str, int size, int *ssize) {
   pMeter = (STabObj *)row;
 
   if (mgmtIsNormalMeter(pMeter)) {
+    if (pMeter->gid.vgId == 0) {
+      return NULL;
+    }
+
     pVgroup = mgmtGetVgroup(pMeter->gid.vgId);
     if (pVgroup == NULL) {
       mError("id:%s not in vgroup:%d", pMeter->meterId, pMeter->gid.vgId);
@@ -426,6 +430,7 @@ void mgmtAddMeterStatisticToAcct(STabObj *pMeter, SAcctObj *pAcct) {
 
 int mgmtInitMeters() {
   void *    pNode = NULL;
+  void *    pLastNode = NULL;
   SVgObj *  pVgroup = NULL;
   STabObj * pMeter = NULL;
   STabObj * pMetric = NULL;
@@ -451,21 +456,47 @@ int mgmtInitMeters() {
 
   pNode = NULL;
   while (1) {
+    pLastNode = pNode;
     pNode = sdbFetchRow(meterSdb, pNode, (void **)&pMeter);
     if (pMeter == NULL) break;
 
     pDb = mgmtGetDbByMeterId(pMeter->meterId);
     if (pDb == NULL) {
-      mError("failed to get db: %s", pMeter->meterId);
+      mError("meter:%s, failed to get db, discard it", pMeter->meterId, pMeter->gid.vgId, pMeter->gid.sid);
+      pMeter->gid.vgId = 0;
+      sdbDeleteRow(meterSdb, pMeter);
+      pNode = pLastNode;
       continue;
     }
 
     if (mgmtIsNormalMeter(pMeter)) {
       pVgroup = mgmtGetVgroup(pMeter->gid.vgId);
-      if (pVgroup == NULL || pVgroup->meterList == NULL) {
-        mError("failed to get vgroup:%i", pMeter->gid.vgId);
+
+      if (pVgroup == NULL) {
+        mError("meter:%s, failed to get vgroup:%d sid:%d, discard it", pMeter->meterId, pMeter->gid.vgId, pMeter->gid.sid);
+        pMeter->gid.vgId = 0;
+        sdbDeleteRow(meterSdb, pMeter);
+        pNode = pLastNode;
         continue;
       }
+
+      if (strcmp(pVgroup->dbName, pDb->name) != 0) {
+        mError("meter:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
+               pMeter->meterId, pDb->name, pMeter->gid.vgId, pVgroup->dbName, pMeter->gid.sid);
+        pMeter->gid.vgId = 0;
+        sdbDeleteRow(meterSdb, pMeter);
+        pNode = pLastNode;
+        continue;
+      }
+
+      if ( pVgroup->meterList == NULL) {
+        mError("meter:%s, vgroup:%d meterlist is null", pMeter->meterId, pMeter->gid.vgId);
+        pMeter->gid.vgId = 0;
+        sdbDeleteRow(meterSdb, pMeter);
+        pNode = pLastNode;
+        continue;
+      }
+
       pVgroup->meterList[pMeter->gid.sid] = pMeter;
       taosIdPoolMarkStatus(pVgroup->idPool, pMeter->gid.sid, 1);