[TD-114] let replica 3 can create table

src/vnode/main/src/vnodeWrite.c

@@ -110,36 +110,27 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
   int32_t code = 0;
 
   dTrace("pVnode:%p vgId:%d, table:%s, start to create", pVnode, pVnode->vgId, pTable->tableId);
-  pTable->numOfColumns  = htons(pTable->numOfColumns);
-  pTable->numOfTags     = htons(pTable->numOfTags);
-  pTable->sid           = htonl(pTable->sid);
-  pTable->sversion      = htonl(pTable->sversion);
-  pTable->tagDataLen    = htonl(pTable->tagDataLen);
-  pTable->sqlDataLen    = htonl(pTable->sqlDataLen);
-  pTable->uid           = htobe64(pTable->uid);
-  pTable->superTableUid = htobe64(pTable->superTableUid);
-  pTable->createdTime   = htobe64(pTable->createdTime);
+  int16_t numOfColumns  = htons(pTable->numOfColumns);
+  int16_t numOfTags     = htons(pTable->numOfTags);
+  int32_t sid           = htonl(pTable->sid);
+  uint64_t uid           = htobe64(pTable->uid);
   SSchema *pSchema = (SSchema *) pTable->data;
 
-  int totalCols = pTable->numOfColumns + pTable->numOfTags;
-  for (int i = 0; i < totalCols; i++) {
-    pSchema[i].colId = htons(pSchema[i].colId);
-    pSchema[i].bytes = htons(pSchema[i].bytes);
-  }
+  int32_t totalCols = numOfColumns + numOfTags;
   
   STableCfg tCfg;
-  tsdbInitTableCfg(&tCfg, pTable->tableType, pTable->uid, pTable->sid);
+  tsdbInitTableCfg(&tCfg, pTable->tableType, uid, sid);
 
-  STSchema *pDestSchema = tdNewSchema(pTable->numOfColumns);
-  for (int i = 0; i < pTable->numOfColumns; i++) {
-    tdSchemaAppendCol(pDestSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
+  STSchema *pDestSchema = tdNewSchema(numOfColumns);
+  for (int i = 0; i < numOfColumns; i++) {
+    tdSchemaAppendCol(pDestSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes));
   }
   tsdbTableSetSchema(&tCfg, pDestSchema, false);
 
-  if (pTable->numOfTags != 0) {
-    STSchema *pDestTagSchema = tdNewSchema(pTable->numOfTags);
-    for (int i = pTable->numOfColumns; i < totalCols; i++) {
-      tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
+  if (numOfTags != 0) {
+    STSchema *pDestTagSchema = tdNewSchema(numOfTags);
+    for (int i = numOfColumns; i < totalCols; i++) {
+      tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes));
     }
     tsdbTableSetTagSchema(&tCfg, pDestTagSchema, false);
 
@@ -147,9 +138,9 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
     int accumBytes = 0;
     SDataRow dataRow = tdNewDataRowFromSchema(pDestTagSchema);
 
-    for (int i = 0; i < pTable->numOfTags; i++) {
+    for (int i = 0; i < numOfTags; i++) {
       tdAppendColVal(dataRow, pTagData + accumBytes, pDestTagSchema->columns + i);
-      accumBytes += pSchema[i + pTable->numOfColumns].bytes;
+      accumBytes += htons(pSchema[i + numOfColumns].bytes);
     }
     tsdbTableSetTagValue(&tCfg, dataRow, false);
   }
@@ -183,51 +174,46 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
   int32_t code = 0;
 
   dTrace("pVnode:%p vgId:%d, table:%s, start to alter", pVnode, pVnode->vgId, pTable->tableId);
-  pTable->numOfColumns  = htons(pTable->numOfColumns);
-  pTable->numOfTags     = htons(pTable->numOfTags);
-  pTable->sid           = htonl(pTable->sid);
-  pTable->sversion      = htonl(pTable->sversion);
-  pTable->tagDataLen    = htonl(pTable->tagDataLen);
-  pTable->sqlDataLen    = htonl(pTable->sqlDataLen);
-  pTable->uid           = htobe64(pTable->uid);
-  pTable->superTableUid = htobe64(pTable->superTableUid);
-  pTable->createdTime   = htobe64(pTable->createdTime);
+  int16_t numOfColumns  = htons(pTable->numOfColumns);
+  int16_t numOfTags     = htons(pTable->numOfTags);
+  int32_t sid           = htonl(pTable->sid);
+  uint64_t uid           = htobe64(pTable->uid);
   SSchema *pSchema = (SSchema *) pTable->data;
 
-  int totalCols = pTable->numOfColumns + pTable->numOfTags;
-  for (int i = 0; i < totalCols; i++) {
-    pSchema[i].colId = htons(pSchema[i].colId);
-    pSchema[i].bytes = htons(pSchema[i].bytes);
-  }
+  int32_t totalCols = numOfColumns + numOfTags;
   
   STableCfg tCfg;
-  tsdbInitTableCfg(&tCfg, pTable->tableType, pTable->uid, pTable->sid);
+  tsdbInitTableCfg(&tCfg, pTable->tableType, uid, sid);
 
-  STSchema *pDestSchema = tdNewSchema(pTable->numOfColumns);
-  for (int i = 0; i < pTable->numOfColumns; i++) {
-    tdSchemaAppendCol(pDestSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
+  STSchema *pDestSchema = tdNewSchema(numOfColumns);
+  for (int i = 0; i < numOfColumns; i++) {
+    tdSchemaAppendCol(pDestSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes));
   }
   tsdbTableSetSchema(&tCfg, pDestSchema, false);
 
-  if (pTable->numOfTags != 0) {
-    STSchema *pDestTagSchema = tdNewSchema(pTable->numOfTags);
-    for (int i = pTable->numOfColumns; i < totalCols; i++) {
-      tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
+  if (numOfTags != 0) {
+    STSchema *pDestTagSchema = tdNewSchema(numOfTags);
+    for (int i = numOfColumns; i < totalCols; i++) {
+      tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes));
     }
-    tsdbTableSetSchema(&tCfg, pDestTagSchema, false);
+    tsdbTableSetTagSchema(&tCfg, pDestTagSchema, false);
 
     char *pTagData = pTable->data + totalCols * sizeof(SSchema);
     int accumBytes = 0;
     SDataRow dataRow = tdNewDataRowFromSchema(pDestTagSchema);
 
-    for (int i = 0; i < pTable->numOfTags; i++) {
+    for (int i = 0; i < numOfTags; i++) {
       tdAppendColVal(dataRow, pTagData + accumBytes, pDestTagSchema->columns + i);
-      accumBytes += pSchema[i + pTable->numOfColumns].bytes;
+      accumBytes += htons(pSchema[i + numOfColumns].bytes);
     }
     tsdbTableSetTagValue(&tCfg, dataRow, false);
   }
 
-  code = tsdbAlterTable(pVnode->tsdb, &tCfg);
+  void *pTsdb = vnodeGetTsdb(pVnode);
+  code = tsdbAlterTable(pTsdb, &tCfg);
+
+  tfree(pDestSchema);
+
   dTrace("pVnode:%p vgId:%d, table:%s, alter table result:%d", pVnode, pVnode->vgId, pTable->tableId, code);
 
   return code;
@@ -238,7 +224,7 @@ static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
   int32_t code = 0;
 
   dTrace("pVnode:%p vgId:%d, stable:%s, start to drop", pVnode, pVnode->vgId, pTable->tableId);
-  pTable->uid = htobe64(pTable->uid);
+  // int64_t uid = htobe64(pTable->uid);
 
   // TODO: drop stable in vvnode
   //void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);