Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/hzcheng_3.0

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

@@ -70,6 +70,7 @@ static int  tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
 static void tsdbResetCommitFile(SCommitH *pCommith);
 static int  tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
 static int  tsdbCommitToTable(SCommitH *pCommith, int tid);
+static bool tsdbCommitIsSameFile(SCommitH *pCommith, int bidx);
 static int  tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx);
 static int  tsdbSetCommitTable(SCommitH *pCommith, STable *pTable);
 static int  tsdbComparKeyBlock(const void *arg1, const void *arg2);
@@ -891,9 +892,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
 }
 
 static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) {
-  SReadH *pReadh = &pCommith->readh;
-  int     nBlocks = pIdx->numOfBlocks;
-  int     bidx = 0;
+  SReadH   *pReadh = &pCommith->readh;
+  STsdb    *pTsdb = TSDB_READ_REPO(pReadh);
+  STSchema *pTSchema = NULL;
+  int       nBlocks = pIdx->numOfBlocks;
+  int       bidx = 0;
 
   tsdbResetCommitTable(pCommith);
 
@@ -903,30 +906,49 @@ static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) {
     return -1;
   }
 
+  STable table = {.tid = pIdx->uid, .uid = pIdx->uid, .pSchema = NULL};
+  pCommith->pTable = &table;
+
   while (bidx < nBlocks) {
+    if (!pTSchema && !tsdbCommitIsSameFile(pCommith, bidx)) {
+      // Set commit table
+      pTSchema = metaGetTbTSchema(REPO_META(pTsdb), pIdx->uid, 0);  // TODO: schema version
+      if (!pTSchema) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
+      }
+      table.pSchema = pTSchema;
+      if (tsdbSetCommitTable(pCommith, &table) < 0) {
+        taosMemoryFreeClear(pTSchema);
+        return -1;
+      }
+    }
+
     if (tsdbMoveBlock(pCommith, bidx) < 0) {
       tsdbError("vgId:%d failed to move block into file %s since %s", TSDB_COMMIT_REPO_ID(pCommith),
                 TSDB_FILE_FULL_NAME(TSDB_COMMIT_HEAD_FILE(pCommith)), tstrerror(terrno));
+      taosMemoryFreeClear(pTSchema);
       return -1;
     }
+
     ++bidx;
   }
 
-  STable table = {.tid = pIdx->uid, .uid = pIdx->uid, .pSchema = NULL};
-  TSDB_COMMIT_TABLE(pCommith) = &table;
-
   if (tsdbWriteBlockInfo(pCommith) < 0) {
     tsdbError("vgId:%d failed to write SBlockInfo part into file %s since %s", TSDB_COMMIT_REPO_ID(pCommith),
               TSDB_FILE_FULL_NAME(TSDB_COMMIT_HEAD_FILE(pCommith)), tstrerror(terrno));
+    taosMemoryFreeClear(pTSchema);
     return -1;
   }
 
+  taosMemoryFreeClear(pTSchema);
   return 0;
 }
 
 static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable) {
   STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
 
+
   pCommith->pTable = pTable;
 
   if (tdInitDataCols(pCommith->pDataCols, pSchema) < 0) {
@@ -1323,6 +1345,14 @@ static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) {
   return 0;
 }
 
+static bool tsdbCommitIsSameFile(SCommitH *pCommith, int bidx) {
+  SBlock *pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
+  if (pBlock->last) {
+    return pCommith->isLFileSame;
+  }
+  return pCommith->isDFileSame;
+}
+
 static int tsdbMoveBlock(SCommitH *pCommith, int bidx) {
   SBlock *pBlock = pCommith->readh.pBlkInfo->blocks + bidx;
   SDFile *pDFile;

source/libs/wal/src/walRead.c

@@ -130,6 +130,7 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
     }
   }
 
+  // code set inner
   if (walReadSeekFilePos(pRead, pRet->firstVer, ver) < 0) {
     return -1;
   }
@@ -249,16 +250,22 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
   // TODO: check wal life
   if (pRead->curVersion != ver) {
     if (walReadSeekVer(pRead, ver) < 0) {
+      terrno = TSDB_CODE_WAL_INVALID_VER;
+      wError("unexpected wal log version: % " PRId64 ", since seek error", ver);
       return -1;
     }
   }
 
-  if (!taosValidFile(pRead->pReadLogTFile)) {
-    return -1;
-  }
+  /*if (!taosValidFile(pRead->pReadLogTFile)) {*/
+  /*return -1;*/
+  /*}*/
 
   code = taosReadFile(pRead->pReadLogTFile, pRead->pHead, sizeof(SWalHead));
   if (code != sizeof(SWalHead)) {
+    if (code < 0)
+      terrno = TAOS_SYSTEM_ERROR(errno);
+    else
+      terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
 

source/libs/wal/src/walWrite.c

@@ -225,6 +225,7 @@ int walRoll(SWal *pWal) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
   }
+  // terrno set inner
   code = walRollFileInfo(pWal);
   if (code != 0) {
     return -1;