fix: use schema version of row when commit

source/dnode/vnode/src/inc/tsdb.h

@@ -40,8 +40,8 @@ typedef struct STable        STable;
 
 int  tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
 void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable);
-int  tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
-                           TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
+int  tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
+                           SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
 
 // tsdbCommit ================
 
@@ -179,7 +179,14 @@ struct STsdbFS {
 int tsdbLockRepo(STsdb *pTsdb);
 int tsdbUnlockRepo(STsdb *pTsdb);
 
-static FORCE_INLINE STSchema *tsdbGetTableSchemaImpl(STable *pTable, bool lock, bool copy, int32_t version) {
+static FORCE_INLINE STSchema *tsdbGetTableSchemaImpl(STsdb *pTsdb, STable *pTable, bool lock, bool copy,
+                                                     int32_t version) {
+
+  if ((version != -1) && (schemaVersion(pTable->pSchema) != version)) {
+    taosMemoryFreeClear(pTable->pSchema);
+    pTable->pSchema = metaGetTbTSchema(REPO_META(pTsdb), pTable->uid, version);
+  }
+
   return pTable->pSchema;
 }
 

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

@@ -84,7 +84,7 @@ static int  tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols
 static void tsdbResetCommitTable(SCommitH *pCommith);
 static void tsdbCloseCommitFile(SCommitH *pCommith, bool hasError);
 static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *pInfo);
-static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
+static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
                                       TSKEY maxKey, int maxRows, int8_t update);
 int         tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
 
@@ -301,7 +301,8 @@ static void tsdbSeekCommitIter(SCommitH *pCommith, TSKEY key) {
     SCommitIter *pIter = pCommith->iters + i;
     if (pIter->pTable == NULL || pIter->pIter == NULL) continue;
 
-    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, key - 1, INT32_MAX, NULL, NULL, 0, true, NULL);
+    tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, pIter->pIter, key - 1, INT32_MAX, NULL, NULL, 0,
+                          true, NULL);
   }
 }
 
@@ -947,7 +948,7 @@ static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) {
 }
 
 static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable) {
-  STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
+  STSchema *pSchema = tsdbGetTableSchemaImpl(TSDB_COMMIT_REPO(pCommith),pTable, false, false, -1);
 
   pCommith->pTable = pTable;
 
@@ -1254,8 +1255,8 @@ static int tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLi
   SBlock     block;
 
   while (true) {
-    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, keyLimit, defaultRows, pCommith->pDataCols, NULL, 0,
-                          pCfg->update, &mInfo);
+    tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, pIter->pIter, keyLimit, defaultRows,
+                          pCommith->pDataCols, NULL, 0, pCfg->update, &mInfo);
 
     if (pCommith->pDataCols->numOfRows <= 0) break;
 
@@ -1298,8 +1299,9 @@ static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) {
   SSkipListIterator titer = *(pIter->pIter);
   if (tsdbLoadBlockDataCols(&(pCommith->readh), pBlock, NULL, &colId, 1, false) < 0) return -1;
 
-  tsdbLoadDataFromCache(pIter->pTable, &titer, keyLimit, INT32_MAX, NULL, pCommith->readh.pDCols[0]->cols[0].pData,
-                        pCommith->readh.pDCols[0]->numOfRows, pCfg->update, &mInfo);
+  tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, &titer, keyLimit, INT32_MAX, NULL,
+                        pCommith->readh.pDCols[0]->cols[0].pData, pCommith->readh.pDCols[0]->numOfRows, pCfg->update,
+                        &mInfo);
 
   if (mInfo.nOperations == 0) {
     // no new data to insert (all updates denied)
@@ -1313,9 +1315,9 @@ static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) {
     *(pIter->pIter) = titer;
   } else if (tsdbCanAddSubBlock(pCommith, pBlock, &mInfo)) {
     // Add a sub-block
-    tsdbLoadDataFromCache(pIter->pTable, pIter->pIter, keyLimit, INT32_MAX, pCommith->pDataCols,
-                          pCommith->readh.pDCols[0]->cols[0].pData, pCommith->readh.pDCols[0]->numOfRows, pCfg->update,
-                          &mInfo);
+    tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, pIter->pIter, keyLimit, INT32_MAX,
+                          pCommith->pDataCols, pCommith->readh.pDCols[0]->cols[0].pData,
+                          pCommith->readh.pDCols[0]->numOfRows, pCfg->update, &mInfo);
     if (pBlock->last) {
       pDFile = TSDB_COMMIT_LAST_FILE(pCommith);
     } else {
@@ -1420,7 +1422,7 @@ static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols
 
   int biter = 0;
   while (true) {
-    tsdbLoadAndMergeFromCache(pCommith->readh.pDCols[0], &biter, pIter, pCommith->pDataCols, keyLimit, defaultRows,
+    tsdbLoadAndMergeFromCache(TSDB_COMMIT_REPO(pCommith), pCommith->readh.pDCols[0], &biter, pIter, pCommith->pDataCols, keyLimit, defaultRows,
                               pCfg->update);
 
     if (pCommith->pDataCols->numOfRows == 0) break;
@@ -1445,7 +1447,7 @@ static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols
   return 0;
 }
 
-static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
+static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
                                       TSKEY maxKey, int maxRows, int8_t update) {
   TSKEY     key1 = INT64_MAX;
   TSKEY     key2 = INT64_MAX;
@@ -1487,7 +1489,7 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
       ++(*iter);
     } else if (key1 > key2) {
       if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
-        pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
+        pSchema = tsdbGetTableSchemaImpl(pTsdb, pCommitIter->pTable, false, false, TD_ROW_SVER(row));
         ASSERT(pSchema != NULL);
       }
 
@@ -1527,7 +1529,7 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
       if (TD_SUPPORT_UPDATE(update)) {
         // copy mem data(Multi-Version)
         if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
-          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
+          pSchema = tsdbGetTableSchemaImpl(pTsdb, pCommitIter->pTable, false, false, TD_ROW_SVER(row));
           ASSERT(pSchema != NULL);
         }
 

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

@@ -20,7 +20,8 @@ static void     tsdbFreeTbData(STbData *pTbData);
 static char    *tsdbGetTsTupleKey(const void *data);
 static int      tsdbTbDataComp(const void *arg1, const void *arg2);
 static char    *tsdbTbDataGetUid(const void *arg);
-static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row, bool merge);
+static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row,
+                                    bool merge);
 
 int tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable) {
   STsdbMemTable *pMemTable;
@@ -88,8 +89,8 @@ void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable) {
  *
  * The function tries to procceed AS MUCH AS POSSIBLE.
  */
-int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
-                          TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo) {
+int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
+                          SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo) {
   ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0);
   if (pIter == NULL) return 0;
   STSchema *pSchema = NULL;
@@ -222,12 +223,12 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
             if (lastKey != TSKEY_INITIAL_VAL) {
               ++pCols->numOfRows;
             }
-            tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
+            tsdbAppendTableRowToCols(pTsdb, pTable, pCols, &pSchema, row, false);
           }
           lastKey = rowKey;
         } else {
           if (keepDup) {
-            tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true);
+            tsdbAppendTableRowToCols(pTsdb, pTable, pCols, &pSchema, row, true);
           } else {
             // discard
           }
@@ -249,7 +250,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
         if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
         pMergeInfo->rowsDeleteSucceed++;
         pMergeInfo->nOperations++;
-        tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
+        tsdbAppendTableRowToCols(pTsdb, pTable, pCols, &pSchema, row, false);
       } else {
         if (keepDup) {
           if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
@@ -262,11 +263,11 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
               if (lastKey != TSKEY_INITIAL_VAL) {
                 ++pCols->numOfRows;
               }
-              tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
+              tsdbAppendTableRowToCols(pTsdb, pTable, pCols, &pSchema, row, false);
             }
             lastKey = rowKey;
           } else {
-            tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true);
+            tsdbAppendTableRowToCols(pTsdb, pTable, pCols, &pSchema, row, true);
           }
         } else {
           pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, fKey);
@@ -321,7 +322,7 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
     return -1;
   }
   strcat(pRsp->tblFName, mr.me.name);
-  
+
   if (mr.me.type == TSDB_NORMAL_TABLE) {
     sverNew = mr.me.ntbEntry.schema.sver;
   } else {
@@ -431,10 +432,12 @@ static char *tsdbTbDataGetUid(const void *arg) {
   STbData *pTbData = (STbData *)arg;
   return (char *)(&(pTbData->uid));
 }
-static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row, bool merge) {
+
+static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row,
+                                    bool merge) {
   if (pCols) {
     if (*ppSchema == NULL || schemaVersion(*ppSchema) != TD_ROW_SVER(row)) {
-      *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, TD_ROW_SVER(row));
+      *ppSchema = tsdbGetTableSchemaImpl(pTsdb, pTable, false, false, TD_ROW_SVER(row));
       if (*ppSchema == NULL) {
         ASSERT(false);
         return -1;

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

@@ -157,7 +157,7 @@ int tsdbLoadBlockIdx(SReadH *pReadh) {
 }
 
 int tsdbSetReadTable(SReadH *pReadh, STable *pTable) {
-  STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
+  STSchema *pSchema = tsdbGetTableSchemaImpl(TSDB_READ_REPO(pReadh), pTable, false, false, -1);
 
   pReadh->pTable = pTable;