From 003002af1dbf2b9b64ddc779f09b3a51ad8575d5 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 16:09:20 +0800
Subject: [PATCH 01/13] enh(query): handle getBufPage return NULL when no
 available disk spaces

---
 include/common/tcommon.h                |   1 +
 source/libs/function/src/builtinsimpl.c | 102 +++++++++++++++++++-----
 2 files changed, 83 insertions(+), 20 deletions(-)

diff --git a/include/common/tcommon.h b/include/common/tcommon.h
index f74795a250..36b1abf48a 100644
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -91,6 +91,7 @@ typedef struct STuplePos {
     };
     STupleKey streamTupleKey;
   };
+  bool isValid;
 } STuplePos;
 
 typedef struct SFirstLastRes {
diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 695f62b7d9..70aa7b4c0b 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -2014,20 +2014,26 @@ static void prepareBuf(SqlFunctionCtx* pCtx) {
   ASSERT(pCtx->subsidiaries.rowLen > 0);
 }
 
-static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx,
+static int32_t firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx,
                                    SFirstLastRes* pInfo) {
   if (pCtx->subsidiaries.num <= 0) {
-    return;
+    return TSDB_CODE_SUCCESS;
   }
 
   if (!pInfo->hasResult) {
     pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock);
+    if (!pInfo->pos.isValid) {
+      terrno = TSDB_CODE_NO_AVAIL_DISK;
+      return terrno;
+    }
   } else {
     updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
-static void doSaveCurrentVal(SqlFunctionCtx* pCtx, int32_t rowIndex, int64_t currentTs, int32_t type, char* pData) {
+static int32_t doSaveCurrentVal(SqlFunctionCtx* pCtx, int32_t rowIndex, int64_t currentTs, int32_t type, char* pData) {
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
   SFirstLastRes*       pInfo = GET_ROWCELL_INTERBUF(pResInfo);
 
@@ -2037,9 +2043,13 @@ static void doSaveCurrentVal(SqlFunctionCtx* pCtx, int32_t rowIndex, int64_t cur
 
   memcpy(pInfo->buf, pData, pInfo->bytes);
   pInfo->ts = currentTs;
-  firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pInfo);
+  int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pInfo);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
 
   pInfo->hasResult = true;
+  return TSDB_CODE_SUCCESS;
 }
 
 // This ordinary first function does not care if current scan is ascending order or descending order scan
@@ -2063,7 +2073,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
   if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
     ASSERT(pInputCol->hasNull == true);
     // save selectivity value for column consisted of all null values
-    firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     return TSDB_CODE_SUCCESS;
   }
 
@@ -2136,7 +2149,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
     char* data = colDataGetData(pInputCol, i);
     TSKEY cts = pts[i];
     if (pResInfo->numOfRes == 0 || pInfo->ts > cts) {
-      doSaveCurrentVal(pCtx, i, cts, pInputCol->info.type, data);
+      int32_t code = doSaveCurrentVal(pCtx, i, cts, pInputCol->info.type, data);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
       pResInfo->numOfRes = 1;
     }
   }
@@ -2144,7 +2160,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
 
   if (numOfElems == 0) {
     // save selectivity value for column consisted of all null values
-    firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
   SET_VAL(pResInfo, numOfElems, 1);
   return TSDB_CODE_SUCCESS;
@@ -2171,7 +2190,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
   if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
     ASSERT(pInputCol->hasNull == true);
     // save selectivity value for column consisted of all null values
-    firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     return TSDB_CODE_SUCCESS;
   }
 
@@ -2261,7 +2283,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
 
       if (pResInfo->numOfRes == 0 || pInfo->ts < cts) {
         char* data = colDataGetData(pInputCol, chosen);
-        doSaveCurrentVal(pCtx, i, cts, type, data);
+        int32_t code = doSaveCurrentVal(pCtx, i, cts, type, data);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
         pResInfo->numOfRes = 1;
       }
     }
@@ -2269,7 +2294,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
     for (int32_t i = pInput->startRowIndex + round * 4; i < pInput->startRowIndex + pInput->numOfRows; ++i) {
       if (pResInfo->numOfRes == 0 || pInfo->ts < pts[i]) {
         char* data = colDataGetData(pInputCol, i);
-        doSaveCurrentVal(pCtx, i, pts[i], type, data);
+        int32_t code = doSaveCurrentVal(pCtx, i, pts[i], type, data);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
         pResInfo->numOfRes = 1;
       }
     }
@@ -2283,7 +2311,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
 
       if (pResInfo->numOfRes == 0 || pInfo->ts < pts[i]) {
         char* data = colDataGetData(pInputCol, i);
-        doSaveCurrentVal(pCtx, i, pts[i], type, data);
+        int32_t code = doSaveCurrentVal(pCtx, i, pts[i], type, data);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
         pResInfo->numOfRes = 1;
       }
     }
@@ -2294,7 +2325,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
 
   // save selectivity value for column consisted of all null values
   if (numOfElems == 0) {
-    firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
 
   //  SET_VAL(pResInfo, numOfElems, 1);
@@ -2322,12 +2356,17 @@ static int32_t firstLastTransferInfoImpl(SFirstLastRes* pInput, SFirstLastRes* p
   return TSDB_CODE_SUCCESS;
 }
 
-static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst,
+static int32_t firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst,
                                   int32_t rowIndex) {
   if (TSDB_CODE_SUCCESS == firstLastTransferInfoImpl(pInput, pOutput, isFirst)) {
-    firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pOutput);
+    int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pOutput);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pOutput->hasResult = true;
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuery) {
@@ -2343,7 +2382,10 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer
   for (int32_t i = start; i < start + pInput->numOfRows; ++i) {
     char*          data = colDataGetData(pCol, i);
     SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data);
-    firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery, i);
+    int32_t code = firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery, i);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     if (!numOfElems) {
       numOfElems = pInputInfo->hasResult ? 1 : 0;
     }
@@ -2412,7 +2454,7 @@ int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void doSaveLastrow(SqlFunctionCtx* pCtx, char* pData, int32_t rowIndex, int64_t cts, SFirstLastRes* pInfo) {
+static int32_t doSaveLastrow(SqlFunctionCtx* pCtx, char* pData, int32_t rowIndex, int64_t cts, SFirstLastRes* pInfo) {
   SInputColumnInfoData* pInput = &pCtx->input;
   SColumnInfoData*      pInputCol = pInput->pData[0];
 
@@ -2429,9 +2471,14 @@ static void doSaveLastrow(SqlFunctionCtx* pCtx, char* pData, int32_t rowIndex, i
   }
 
   pInfo->ts = cts;
-  firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pInfo);
+  int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pInfo);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
 
   pInfo->hasResult = true;
+
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
@@ -2493,7 +2540,10 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
 
     numOfElems++;
     if (pResInfo->numOfRes == 0 || pInfo->ts < cts) {
-      doSaveLastrow(pCtx, data, i, cts, pInfo);
+      int32_t code = doSaveLastrow(pCtx, data, i, cts, pInfo);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
       pResInfo->numOfRes = 1;
     }
   }
@@ -2948,18 +2998,27 @@ static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf
 
     if (pHandle->currentPage == -1) {
       pPage = getNewBufPage(pHandle->pBuf, &pHandle->currentPage);
+      if (pPage == NULL) {
+        return p;
+      }
       pPage->num = sizeof(SFilePage);
     } else {
       pPage = getBufPage(pHandle->pBuf, pHandle->currentPage);
+      if (pPage == NULL) {
+        return p;
+      }
       if (pPage->num + length > getBufPageSize(pHandle->pBuf)) {
         // current page is all used, let's prepare a new buffer page
         releaseBufPage(pHandle->pBuf, pPage);
         pPage = getNewBufPage(pHandle->pBuf, &pHandle->currentPage);
+        if (pPage == NULL) {
+          return p;
+        }
         pPage->num = sizeof(SFilePage);
       }
     }
 
-    p = (STuplePos){.pageId = pHandle->currentPage, .offset = pPage->num};
+    p = (STuplePos){.pageId = pHandle->currentPage, .offset = pPage->num, .isValid = true};
     memcpy(pPage->data + pPage->num, pBuf, length);
 
     pPage->num += length;
@@ -5697,7 +5756,10 @@ int32_t cachedLastRowFunction(SqlFunctionCtx* pCtx) {
 
     TSKEY cts = getRowPTs(pInput->pPTS, i);
     if (pResInfo->numOfRes == 0 || pInfo->ts < cts) {
-      doSaveLastrow(pCtx, data, i, cts, pInfo);
+      int32_t code = doSaveLastrow(pCtx, data, i, cts, pInfo);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
       pResInfo->numOfRes = 1;
     }
   }

From d5c92d37880a939fab56ea7c8b839c09ec8a641c Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 16:59:55 +0800
Subject: [PATCH 02/13] fix saveTuplePos

---
 include/common/tcommon.h                  |   1 -
 source/libs/function/inc/builtinsimpl.h   |   5 +-
 source/libs/function/src/builtinsimpl.c   | 111 ++++++++++++++++------
 source/libs/function/src/detail/tminmax.c |  40 ++++++--
 4 files changed, 115 insertions(+), 42 deletions(-)

diff --git a/include/common/tcommon.h b/include/common/tcommon.h
index 36b1abf48a..f74795a250 100644
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -91,7 +91,6 @@ typedef struct STuplePos {
     };
     STupleKey streamTupleKey;
   };
-  bool isValid;
 } STuplePos;
 
 typedef struct SFirstLastRes {
diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h
index cbda8dc472..dc884a0581 100644
--- a/source/libs/function/inc/builtinsimpl.h
+++ b/source/libs/function/inc/builtinsimpl.h
@@ -44,9 +44,10 @@ typedef struct SMinmaxResInfo {
   bool      nullTupleSaved;
   int16_t   type;
 } SMinmaxResInfo;
-int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc);
 
-STuplePos   saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock);
+int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc, int32_t* nElems);
+
+int32_t     saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
 int32_t     updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
 const char* loadTupleData(SqlFunctionCtx* pCtx, const STuplePos* pPos);
 
diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 70aa7b4c0b..8f099d7275 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -757,13 +757,21 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
 }
 
 int32_t minFunction(SqlFunctionCtx* pCtx) {
-  int32_t numOfElems = doMinMaxHelper(pCtx, 1);
+  int32_t numOfElems = 0;
+  int32_t code = doMinMaxHelper(pCtx, 1, &numOfElems);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
   SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
   return TSDB_CODE_SUCCESS;
 }
 
 int32_t maxFunction(SqlFunctionCtx* pCtx) {
-  int32_t numOfElems = doMinMaxHelper(pCtx, 0);
+  int32_t numOfElems = 0;
+  int32_t code = doMinMaxHelper(pCtx, 0, &numOfElems);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
   SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
   return TSDB_CODE_SUCCESS;
 }
@@ -2021,10 +2029,9 @@ static int32_t firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowI
   }
 
   if (!pInfo->hasResult) {
-    pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock);
-    if (!pInfo->pos.isValid) {
-      terrno = TSDB_CODE_NO_AVAIL_DISK;
-      return terrno;
+    int32_t code = saveTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
     }
   } else {
     updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
@@ -2804,8 +2811,8 @@ static STopBotRes* getTopBotOutputInfo(SqlFunctionCtx* pCtx) {
   return pRes;
 }
 
-static void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
-                            uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery);
+static int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
+                               uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery);
 
 static void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type, bool isTopQuery);
 
@@ -2827,11 +2834,17 @@ int32_t topFunction(SqlFunctionCtx* pCtx) {
 
     numOfElems++;
     char* data = colDataGetData(pCol, i);
-    doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, true);
+    int32_t code = doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, true);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
 
   if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
-    pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
+    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pRes->nullTupleSaved = true;
   }
   return TSDB_CODE_SUCCESS;
@@ -2855,11 +2868,17 @@ int32_t bottomFunction(SqlFunctionCtx* pCtx) {
 
     numOfElems++;
     char* data = colDataGetData(pCol, i);
-    doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, false);
+    int32_t code = doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, false);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
 
   if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
-    pRes->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
+    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pRes->nullTupleSaved = true;
   }
 
@@ -2899,7 +2918,7 @@ static int32_t topBotResComparFn(const void* p1, const void* p2, const void* par
   return (val1->v.d > val2->v.d) ? 1 : -1;
 }
 
-void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
+int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
                      uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery) {
   STopBotRes* pRes = getTopBotOutputInfo(pCtx);
 
@@ -2917,7 +2936,10 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
 
     // save the data of this tuple
     if (pCtx->subsidiaries.num > 0) {
-      pItem->tuplePos = saveTupleData(pCtx, rowIndex, pSrcBlock);
+      int32_t code = saveTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
 #ifdef BUF_PAGE_DEBUG
     qDebug("page_saveTuple i:%d, item:%p,pageId:%d, offset:%d\n", pEntryInfo->numOfRes, pItem, pItem->tuplePos.pageId,
@@ -2952,6 +2974,8 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
                      topBotResComparFn, NULL, !isTopQuery);
     }
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 /*
@@ -2991,7 +3015,8 @@ void* serializeTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SSubsid
   return buf;
 }
 
-static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STupleKey key) {
+static int32_t doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STupleKey key,
+                               STuplePos* pPos) {
   STuplePos p = {0};
   if (pHandle->pBuf != NULL) {
     SFilePage* pPage = NULL;
@@ -2999,26 +3024,29 @@ static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf
     if (pHandle->currentPage == -1) {
       pPage = getNewBufPage(pHandle->pBuf, &pHandle->currentPage);
       if (pPage == NULL) {
-        return p;
+        terrno = TSDB_CODE_NO_AVAIL_DISK;
+        return terrno;
       }
       pPage->num = sizeof(SFilePage);
     } else {
       pPage = getBufPage(pHandle->pBuf, pHandle->currentPage);
       if (pPage == NULL) {
-        return p;
+        terrno = TSDB_CODE_NO_AVAIL_DISK;
+        return terrno;
       }
       if (pPage->num + length > getBufPageSize(pHandle->pBuf)) {
         // current page is all used, let's prepare a new buffer page
         releaseBufPage(pHandle->pBuf, pPage);
         pPage = getNewBufPage(pHandle->pBuf, &pHandle->currentPage);
         if (pPage == NULL) {
-          return p;
+          terrno = TSDB_CODE_NO_AVAIL_DISK;
+          return terrno;
         }
         pPage->num = sizeof(SFilePage);
       }
     }
 
-    p = (STuplePos){.pageId = pHandle->currentPage, .offset = pPage->num, .isValid = true};
+    p = (STuplePos){.pageId = pHandle->currentPage, .offset = pPage->num};
     memcpy(pPage->data + pPage->num, pBuf, length);
 
     pPage->num += length;
@@ -3032,10 +3060,11 @@ static STuplePos doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf
     p.streamTupleKey = key;
   }
 
-  return p;
+  *pPos = p;
+  return TSDB_CODE_SUCCESS;
 }
 
-STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock) {
+int32_t saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos) {
   prepareBuf(pCtx);
 
   STupleKey key;
@@ -3050,7 +3079,7 @@ STuplePos saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBloc
   }
 
   char* buf = serializeTupleData(pSrcBlock, rowIndex, &pCtx->subsidiaries, pCtx->subsidiaries.buf);
-  return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, key);
+  return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, key, pPos);
 }
 
 static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STuplePos* pPos) {
@@ -4490,12 +4519,15 @@ static void sampleAssignResult(SSampleInfo* pInfo, char* data, int32_t index) {
   assignVal(pInfo->data + index * pInfo->colBytes, data, pInfo->colBytes, pInfo->colType);
 }
 
-static void doReservoirSample(SqlFunctionCtx* pCtx, SSampleInfo* pInfo, char* data, int32_t index) {
+static int32_t doReservoirSample(SqlFunctionCtx* pCtx, SSampleInfo* pInfo, char* data, int32_t index) {
   pInfo->totalPoints++;
   if (pInfo->numSampled < pInfo->samples) {
     sampleAssignResult(pInfo, data, pInfo->numSampled);
     if (pCtx->subsidiaries.num > 0) {
-      pInfo->tuplePos[pInfo->numSampled] = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+      int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pInfo->tuplePos[pInfo->numSampled]);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
     pInfo->numSampled++;
   } else {
@@ -4507,6 +4539,8 @@ static void doReservoirSample(SqlFunctionCtx* pCtx, SSampleInfo* pInfo, char* da
       }
     }
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t sampleFunction(SqlFunctionCtx* pCtx) {
@@ -4522,11 +4556,17 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
     }
 
     char* data = colDataGetData(pInputCol, i);
-    doReservoirSample(pCtx, pInfo, data, i);
+    int32_t code = doReservoirSample(pCtx, pInfo, data, i);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
 
   if (pInfo->numSampled == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
-    pInfo->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
+    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pInfo->nullTuplePos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pInfo->nullTupleSaved = true;
   }
 
@@ -4820,7 +4860,7 @@ bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
   return true;
 }
 
-static void doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCtx, char* data) {
+static int32_t doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCtx, char* data) {
   int32_t     hashKeyBytes = IS_STR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
   SModeItem** pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
   if (pHashItem == NULL) {
@@ -4830,7 +4870,10 @@ static void doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCtx,
     pItem->count += 1;
 
     if (pCtx->subsidiaries.num > 0) {
-      pItem->tuplePos = saveTupleData(pCtx, rowIndex, pCtx->pSrcBlock);
+      int32_t code = saveTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &pItem->tuplePos);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
 
     taosHashPut(pInfo->pHash, data, hashKeyBytes, &pItem, sizeof(SModeItem*));
@@ -4841,6 +4884,8 @@ static void doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCtx,
       updateTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &((*pHashItem)->tuplePos));
     }
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t modeFunction(SqlFunctionCtx* pCtx) {
@@ -4861,7 +4906,10 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) {
     numOfElems++;
 
     char* data = colDataGetData(pInputCol, i);
-    doModeAdd(pInfo, i, pCtx, data);
+    int32_t code = doModeAdd(pInfo, i, pCtx, data);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
 
     if (sizeof(SModeInfo) + pInfo->numOfPoints * (sizeof(SModeItem) + pInfo->colBytes) >= MODE_MAX_RESULT_SIZE) {
       taosHashCleanup(pInfo->pHash);
@@ -4870,7 +4918,10 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) {
   }
 
   if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
-    pInfo->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
+    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pInfo->nullTuplePos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pInfo->nullTupleSaved = true;
   }
 
diff --git a/source/libs/function/src/detail/tminmax.c b/source/libs/function/src/detail/tminmax.c
index 3660ec272f..cb66fd36fb 100644
--- a/source/libs/function/src/detail/tminmax.c
+++ b/source/libs/function/src/detail/tminmax.c
@@ -700,7 +700,7 @@ static void doExtractVal(SColumnInfoData* pCol, int32_t i, int32_t end, SqlFunct
   }
 }
 
-int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
+int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc, int32_t* nElems) {
   int32_t numOfElems = 0;
 
   SInputColumnInfoData* pInput = &pCtx->input;
@@ -746,7 +746,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
       if (pCtx->subsidiaries.num > 0) {
         index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
         if (index >= 0) {
-          pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+          int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
+          if (code != TSDB_CODE_SUCCESS) {
+            return code;
+          }
         }
       }
     } else {
@@ -760,7 +763,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           if (pCtx->subsidiaries.num > 0) {
             index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
             if (index >= 0) {
-              pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+              int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
+              if (code != TSDB_CODE_SUCCESS) {
+                return code;
+              }
             }
           }
         }
@@ -774,7 +780,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           if (pCtx->subsidiaries.num > 0) {
             index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
             if (index >= 0) {
-              pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+              int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
+              if (code != TSDB_CODE_SUCCESS) {
+                return code;
+              }
             }
           }
         }
@@ -788,7 +797,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           if (pCtx->subsidiaries.num > 0) {
             index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
             if (index >= 0) {
-              pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+              int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
+              if (code != TSDB_CODE_SUCCESS) {
+                return code;
+              }
             }
           }
         }
@@ -804,7 +816,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
         if (pCtx->subsidiaries.num > 0) {
           index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
           if (index >= 0) {
-            pBuf->tuplePos = saveTupleData(pCtx, index, pCtx->pSrcBlock);
+            int32_t code = saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
+            if (code != TSDB_CODE_SUCCESS) {
+              return code;
+            }
           }
         }
       }
@@ -825,7 +840,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
       memcpy(&pBuf->v, pCol->pData + (pCol->info.bytes * i), pCol->info.bytes);
 
       if (pCtx->subsidiaries.num > 0) {
-        pBuf->tuplePos = saveTupleData(pCtx, i, pCtx->pSrcBlock);
+        int32_t code = saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
       }
       pBuf->assign = true;
       numOfElems = 1;
@@ -889,9 +907,13 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
 
 _over:
   if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved) {
-    pBuf->nullTuplePos = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock);
+    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pBuf->nullTuplePos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
     pBuf->nullTupleSaved = true;
   }
 
-  return numOfElems;
+  *nElems = numOfElems;
+  return TSDB_CODE_SUCCESS;
 }

From 6411e717df94c80d61703795c1033217badd4d18 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 17:18:14 +0800
Subject: [PATCH 03/13] fix updateTupleData

---
 source/libs/function/src/builtinsimpl.c | 33 ++++++++++++++++---------
 1 file changed, 22 insertions(+), 11 deletions(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 8f099d7275..5152f7b42d 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -2024,20 +2024,19 @@ static void prepareBuf(SqlFunctionCtx* pCtx) {
 
 static int32_t firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx,
                                    SFirstLastRes* pInfo) {
+  int32_t code = TSDB_CODE_SUCCESS;
+
   if (pCtx->subsidiaries.num <= 0) {
     return TSDB_CODE_SUCCESS;
   }
 
   if (!pInfo->hasResult) {
-    int32_t code = saveTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
+    code = saveTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
   } else {
-    updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
+    code = updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
   }
 
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
 
 static int32_t doSaveCurrentVal(SqlFunctionCtx* pCtx, int32_t rowIndex, int64_t currentTs, int32_t type, char* pData) {
@@ -2965,7 +2964,10 @@ int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSD
 
       // save the data of this tuple by over writing the old data
       if (pCtx->subsidiaries.num > 0) {
-        updateTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
+        int32_t code = updateTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
       }
 #ifdef BUF_PAGE_DEBUG
       qDebug("page_copyTuple pageId:%d, offset:%d", pItem->tuplePos.pageId, pItem->tuplePos.offset);
@@ -3085,6 +3087,10 @@ int32_t saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock*
 static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STuplePos* pPos) {
   if (pHandle->pBuf != NULL) {
     SFilePage* pPage = getBufPage(pHandle->pBuf, pPos->pageId);
+    if (pPage == NULL) {
+      terrno = TSDB_CODE_NO_AVAIL_DISK;
+      return terrno;
+    }
     memcpy(pPage->data + pPos->offset, pBuf, length);
     setBufPageDirty(pPage, true);
     releaseBufPage(pHandle->pBuf, pPage);
@@ -3099,8 +3105,7 @@ int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBloc
   prepareBuf(pCtx);
 
   char* buf = serializeTupleData(pSrcBlock, rowIndex, &pCtx->subsidiaries, pCtx->subsidiaries.buf);
-  doUpdateTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, pPos);
-  return TSDB_CODE_SUCCESS;
+  return doUpdateTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, pPos);
 }
 
 static char* doLoadTupleData(SSerializeDataHandle* pHandle, const STuplePos* pPos) {
@@ -4535,7 +4540,10 @@ static int32_t doReservoirSample(SqlFunctionCtx* pCtx, SSampleInfo* pInfo, char*
     if (j < pInfo->samples) {
       sampleAssignResult(pInfo, data, j);
       if (pCtx->subsidiaries.num > 0) {
-        updateTupleData(pCtx, index, pCtx->pSrcBlock, &pInfo->tuplePos[j]);
+        int32_t code = updateTupleData(pCtx, index, pCtx->pSrcBlock, &pInfo->tuplePos[j]);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
       }
     }
   }
@@ -4881,7 +4889,10 @@ static int32_t doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCt
   } else {
     (*pHashItem)->count += 1;
     if (pCtx->subsidiaries.num > 0) {
-      updateTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &((*pHashItem)->tuplePos));
+      int32_t code = updateTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &((*pHashItem)->tuplePos));
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
   }
 

From af0bd9c72bf6ee40a9650174f63536c7f1f08368 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 17:59:30 +0800
Subject: [PATCH 04/13] fix loadTupleData

---
 source/libs/function/src/builtinsimpl.c | 69 ++++++++++++++++---------
 1 file changed, 44 insertions(+), 25 deletions(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 5152f7b42d..4ebe7404a7 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -776,13 +776,14 @@ int32_t maxFunction(SqlFunctionCtx* pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex);
-static void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos,
+static int32_t setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex);
+static int32_t setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos,
                                 int32_t rowIndex);
 
 int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
-  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
+  int32_t code = TSDB_CODE_SUCCESS;
 
+  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
   SMinmaxResInfo* pRes = GET_ROWCELL_INTERBUF(pEntryInfo);
 
   int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
@@ -799,17 +800,17 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   }
 
   if (pEntryInfo->numOfRes > 0) {
-    setSelectivityValue(pCtx, pBlock, &pRes->tuplePos, currentRow);
+    code = setSelectivityValue(pCtx, pBlock, &pRes->tuplePos, currentRow);
   } else {
-    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
+    code = setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
   }
 
-  return pEntryInfo->numOfRes;
+  return code;
 }
 
-void setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex) {
+int32_t setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex) {
   if (pCtx->subsidiaries.num <= 0) {
-    return;
+    return TSDB_CODE_SUCCESS;
   }
 
   for (int32_t j = 0; j < pCtx->subsidiaries.num; ++j) {
@@ -819,17 +820,23 @@ void setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t
     SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
     colDataAppendNULL(pDstCol, rowIndex);
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
-void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos, int32_t rowIndex) {
+int32_t setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos, int32_t rowIndex) {
   if (pCtx->subsidiaries.num <= 0) {
-    return;
+    return TSDB_CODE_SUCCESS;
   }
 
   if ((pCtx->saveHandle.pBuf != NULL && pTuplePos->pageId != -1) ||
       (pCtx->saveHandle.pState && pTuplePos->streamTupleKey.ts > 0)) {
     int32_t     numOfCols = pCtx->subsidiaries.num;
     const char* p = loadTupleData(pCtx, pTuplePos);
+    if (p == NULL) {
+      terrno = TSDB_CODE_NO_AVAIL_DISK;
+      return terrno;
+    }
 
     bool* nullList = (bool*)p;
     char* pStart = (char*)(nullList + numOfCols * sizeof(bool));
@@ -853,6 +860,8 @@ void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuple
       tdbFree((void*)p);
     }
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 void releaseSource(STuplePos* pPos) {
@@ -2406,6 +2415,7 @@ int32_t firstFunctionMerge(SqlFunctionCtx* pCtx) { return firstLastFunctionMerge
 int32_t lastFunctionMerge(SqlFunctionCtx* pCtx) { return firstLastFunctionMergeImpl(pCtx, false); }
 
 int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t          code = TSDB_CODE_SUCCESS;
   int32_t          slotId = pCtx->pExpr->base.resSchema.slotId;
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
@@ -2416,12 +2426,14 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   colDataAppend(pCol, pBlock->info.rows, pRes->buf, pRes->isNull || pResInfo->isNullRes);
 
   // handle selectivity
-  setSelectivityValue(pCtx, pBlock, &pRes->pos, pBlock->info.rows);
+  code = setSelectivityValue(pCtx, pBlock, &pRes->pos, pBlock->info.rows);
 
-  return pResInfo->numOfRes;
+  return code;
 }
 
 int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t code = TSDB_CODE_SUCCESS;
+
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
   SFirstLastRes*       pRes = GET_ROWCELL_INTERBUF(pEntryInfo);
 
@@ -2437,10 +2449,10 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   colDataAppend(pCol, pBlock->info.rows, res, false);
-  setSelectivityValue(pCtx, pBlock, &pRes->pos, pBlock->info.rows);
+  code = setSelectivityValue(pCtx, pBlock, &pRes->pos, pBlock->info.rows);
 
   taosMemoryFree(res);
-  return 1;
+  return code;
 }
 
 int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
@@ -3111,6 +3123,9 @@ int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBloc
 static char* doLoadTupleData(SSerializeDataHandle* pHandle, const STuplePos* pPos) {
   if (pHandle->pBuf != NULL) {
     SFilePage* pPage = getBufPage(pHandle->pBuf, pPos->pageId);
+    if (pPage == NULL) {
+      return NULL;
+    }
     char*      p = pPage->data + pPos->offset;
     releaseBufPage(pHandle->pBuf, pPage);
     return p;
@@ -3127,6 +3142,8 @@ const char* loadTupleData(SqlFunctionCtx* pCtx, const STuplePos* pPos) {
 }
 
 int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t code = TSDB_CODE_SUCCESS;
+
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
   STopBotRes*          pRes = getTopBotOutputInfo(pCtx);
 
@@ -3139,8 +3156,8 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   int32_t currentRow = pBlock->info.rows;
   if (pEntryInfo->numOfRes <= 0) {
     colDataAppendNULL(pCol, currentRow);
-    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
-    return pEntryInfo->numOfRes;
+    code = setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
+    return code;
   }
   for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
     STopBotResItem* pItem = &pRes->pItems[i];
@@ -3149,11 +3166,11 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
     qDebug("page_finalize i:%d,item:%p,pageId:%d, offset:%d\n", i, pItem, pItem->tuplePos.pageId,
            pItem->tuplePos.offset);
 #endif
-    setSelectivityValue(pCtx, pBlock, &pRes->pItems[i].tuplePos, currentRow);
+    code = setSelectivityValue(pCtx, pBlock, &pRes->pItems[i].tuplePos, currentRow);
     currentRow += 1;
   }
 
-  return pEntryInfo->numOfRes;
+  return code;
 }
 
 void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type, bool isTopQuery) {
@@ -4583,6 +4600,7 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
 }
 
 int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t code = TSDB_CODE_SUCCESS;
   SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
 
   SSampleInfo* pInfo = getSampleOutputInfo(pCtx);
@@ -4594,15 +4612,15 @@ int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   int32_t currentRow = pBlock->info.rows;
   if (pInfo->numSampled == 0) {
     colDataAppendNULL(pCol, currentRow);
-    setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
-    return pInfo->numSampled;
+    code = setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
+    return code;
   }
   for (int32_t i = 0; i < pInfo->numSampled; ++i) {
     colDataAppend(pCol, currentRow + i, pInfo->data + i * pInfo->colBytes, false);
-    setSelectivityValue(pCtx, pBlock, &pInfo->tuplePos[i], currentRow + i);
+    code = setSelectivityValue(pCtx, pBlock, &pInfo->tuplePos[i], currentRow + i);
   }
 
-  return pInfo->numSampled;
+  return code;
 }
 
 bool getTailFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
@@ -4942,6 +4960,7 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) {
 }
 
 int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t              code = TSDB_CODE_SUCCESS;
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
   SModeInfo*           pInfo = GET_ROWCELL_INTERBUF(pResInfo);
   int32_t              slotId = pCtx->pExpr->base.resSchema.slotId;
@@ -4961,15 +4980,15 @@ int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   if (maxCount != 0) {
     SModeItem* pResItem = (SModeItem*)(pInfo->pItems + resIndex * (sizeof(SModeItem) + pInfo->colBytes));
     colDataAppend(pCol, currentRow, pResItem->data, false);
-    setSelectivityValue(pCtx, pBlock, &pResItem->tuplePos, currentRow);
+    code = setSelectivityValue(pCtx, pBlock, &pResItem->tuplePos, currentRow);
   } else {
     colDataAppendNULL(pCol, currentRow);
-    setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
+    code = setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
   }
 
   taosHashCleanup(pInfo->pHash);
 
-  return pResInfo->numOfRes;
+  return code;
 }
 
 bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) {

From c28ba32c93785938cb866503efab20480b573c8f Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Wed, 4 Jan 2023 18:08:33 +0800
Subject: [PATCH 05/13] fix(vnode/mgmt): pre close vnode after queue's empty

---
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 6 +++---
 source/dnode/mnode/impl/src/mndDnode.c   | 2 +-
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index f808c67ef6..9e91670d33 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -84,9 +84,6 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   taosThreadRwlockUnlock(&pMgmt->lock);
   vmReleaseVnode(pMgmt, pVnode);
 
-  dInfo("vgId:%d, pre close", pVnode->vgId);
-  vnodePreClose(pVnode->pImpl);
-
   dInfo("vgId:%d, wait for vnode ref become 0", pVnode->vgId);
   while (pVnode->refCount > 0) taosMsleep(10);
 
@@ -118,6 +115,9 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   dInfo("vgId:%d, all vnode queues is empty", pVnode->vgId);
 
+  dInfo("vgId:%d, pre close", pVnode->vgId);
+  vnodePreClose(pVnode->pImpl);
+
   vmFreeQueue(pMgmt, pVnode);
   vnodeClose(pVnode->pImpl);
   pVnode->pImpl = NULL;
diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index d7b16c2c8e..c7a416d444 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -1050,7 +1050,7 @@ static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
         status = "offline";
     }
 
-    char b1[9] = {0};
+    char b1[16] = {0};
     STR_TO_VARSTR(b1, status);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, b1, false);

From 929bafa26476d96cd3b5c1a944b35897a259011b Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 18:15:16 +0800
Subject: [PATCH 06/13] fix getBufPage return NULL in tpercentile.c

---
 source/libs/function/src/builtinsimpl.c |  5 +++++
 source/libs/function/src/tpercentile.c  | 16 +++++++++++++++-
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 4ebe7404a7..31ffe9365c 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -1673,6 +1673,11 @@ int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   }
 
   tMemBucketDestroy(pMemBucket);
+
+  if (ppInfo->result < 0) {
+    return TSDB_CODE_NO_AVAIL_DISK;
+  }
+
   return functionFinalize(pCtx, pBlock);
 }
 
diff --git a/source/libs/function/src/tpercentile.c b/source/libs/function/src/tpercentile.c
index 157ee08f15..ef81a8fca5 100644
--- a/source/libs/function/src/tpercentile.c
+++ b/source/libs/function/src/tpercentile.c
@@ -40,6 +40,9 @@ static SFilePage *loadDataFromFilePage(tMemBucket *pMemBucket, int32_t slotIdx)
     int32_t *pageId = taosArrayGet(pIdList, i);
 
     SFilePage *pg = getBufPage(pMemBucket->pBuffer, *pageId);
+    if (pg == NULL) {
+      return NULL;
+    }
     memcpy(buffer->data + offset, pg->data, (size_t)(pg->num * pMemBucket->bytes));
 
     offset += (int32_t)(pg->num * pMemBucket->bytes);
@@ -104,6 +107,9 @@ double findOnlyResult(tMemBucket *pMemBucket) {
 
       int32_t   *pageId = taosArrayGet(list, 0);
       SFilePage *pPage = getBufPage(pMemBucket->pBuffer, *pageId);
+      if (pPage == NULL) {
+        return -1;
+      }
       assert(pPage->num == 1);
 
       double v = 0;
@@ -470,6 +476,9 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
       if (pSlot->info.size <= pMemBucket->maxCapacity) {
         // data in buffer and file are merged together to be processed.
         SFilePage *buffer = loadDataFromFilePage(pMemBucket, i);
+        if (buffer == NULL) {
+          return -1;
+        }
         int32_t    currentIdx = count - num;
 
         char *thisVal = buffer->data + pMemBucket->bytes * currentIdx;
@@ -504,6 +513,9 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
         for (int32_t f = 0; f < list->size; ++f) {
           int32_t *pageId = taosArrayGet(list, f);
           SFilePage *pg = getBufPage(pMemBucket->pBuffer, *pageId);
+          if (pg == NULL) {
+            return -1;
+          }
 
           tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
           setBufPageDirty(pg, true);
@@ -527,7 +539,9 @@ double getPercentile(tMemBucket *pMemBucket, double percent) {
 
   // if only one elements exists, return it
   if (pMemBucket->total == 1) {
-    return findOnlyResult(pMemBucket);
+    if (findOnlyResult(pMemBucket) < 0) {
+      return -1;
+    }
   }
 
   percent = fabs(percent);

From 898955521d0003a0b331dcb67390dd18ce3a4918 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 4 Jan 2023 18:15:16 +0800
Subject: [PATCH 07/13] fix getBufPage return NULL in tpercentile.c

---
 source/libs/function/src/builtinsimpl.c |  7 +++++--
 source/libs/function/src/tpercentile.c  | 10 ++++++++--
 2 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 31ffe9365c..f6ebaace5b 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -1587,7 +1587,7 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
     // all data are null, set it completed
     if (pInfo->numOfElems == 0) {
       pResInfo->complete = true;
-      return 0;
+      return TSDB_CODE_SUCCESS;
     } else {
       pInfo->pMemBucket = tMemBucketCreate(pCol->info.bytes, type, pInfo->minval, pInfo->maxval);
     }
@@ -1650,7 +1650,10 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
 
       char* data = colDataGetData(pCol, i);
       numOfElems += 1;
-      tMemBucketPut(pInfo->pMemBucket, data, 1);
+      int32_t code = tMemBucketPut(pInfo->pMemBucket, data, 1);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
 
     SET_VAL(pResInfo, numOfElems, 1);
diff --git a/source/libs/function/src/tpercentile.c b/source/libs/function/src/tpercentile.c
index ef81a8fca5..d5983ae469 100644
--- a/source/libs/function/src/tpercentile.c
+++ b/source/libs/function/src/tpercentile.c
@@ -385,6 +385,9 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
       }
 
       pSlot->info.data = getNewBufPage(pBucket->pBuffer, &pageId);
+      if (pSlot->info.data == NULL) {
+        return TSDB_CODE_NO_AVAIL_DISK;
+      }
       pSlot->info.pageId = pageId;
       taosArrayPush(pPageIdList, &pageId);
     }
@@ -396,7 +399,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
   }
 
   pBucket->total += count;
-  return 0;
+  return TSDB_CODE_SUCCESS;
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////
@@ -517,7 +520,10 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
             return -1;
           }
 
-          tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
+          int32_t code = tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
+          if (code != TSDB_CODE_SUCCESS) {
+            return -1;
+          }
           setBufPageDirty(pg, true);
           releaseBufPage(pMemBucket->pBuffer, pg);
         }

From ffd901fa6f77553b122b1177b1aedeeb27da0ed1 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Thu, 5 Jan 2023 09:13:57 +0800
Subject: [PATCH 08/13] fix mem leak

---
 source/libs/function/src/builtinsimpl.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index dfd0430d90..8fde27e046 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -1649,6 +1649,7 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
       numOfElems += 1;
       int32_t code = tMemBucketPut(pInfo->pMemBucket, data, 1);
       if (code != TSDB_CODE_SUCCESS) {
+        tMemBucketDestroy(pInfo->pMemBucket);
         return code;
       }
     }

From 183b0002404aad0184d778794e06f096806022b5 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Thu, 5 Jan 2023 10:18:18 +0800
Subject: [PATCH 09/13] fix error

---
 source/libs/function/src/detail/tminmax.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/function/src/detail/tminmax.c b/source/libs/function/src/detail/tminmax.c
index 1a46368874..847c738655 100644
--- a/source/libs/function/src/detail/tminmax.c
+++ b/source/libs/function/src/detail/tminmax.c
@@ -825,7 +825,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc, int32_t* nElems)
     }
 
     pBuf->assign = true;
-    return numOfElems;
+    return TSDB_CODE_SUCCESS;
   }
 
   int32_t start = pInput->startRowIndex;

From 5845c9c87eb58674b68a1034dadd7cf4f981f3c4 Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Thu, 5 Jan 2023 10:29:31 +0800
Subject: [PATCH 10/13] ci:add ci for keepColumnName

---
 tests/script/tsim/stream/basic3.sim | 75 ++++++++++++++++++++++++++++-
 1 file changed, 73 insertions(+), 2 deletions(-)

diff --git a/tests/script/tsim/stream/basic3.sim b/tests/script/tsim/stream/basic3.sim
index 41e19b19af..e598919e34 100644
--- a/tests/script/tsim/stream/basic3.sim
+++ b/tests/script/tsim/stream/basic3.sim
@@ -1,6 +1,7 @@
 system sh/stop_dnodes.sh
 system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c debugflag 131
+system sh/cfg.sh -n dnode1 -c debugflag -v 131
+system sh/cfg.sh -n dnode1 -c keepColumnName -v 1
 system sh/exec.sh -n dnode1 -s start
 
 sleep 5000
@@ -9,7 +10,7 @@ sql connect
 
 print ========== interval\session\state window
 
-sql CREATE DATABASE test1 BUFFER 96 CACHESIZE 1 CACHEMODEL 'none' COMP 2 DURATION 14400m WAL_FSYNC_PERIOD 3000 MAXROWS 4096 MINROWS 100 KEEP 5256000m,5256000m,5256000m PAGES 256 PAGESIZE 4 PRECISION 'ms' REPLICA 1 STRICT 'off' WAL_LEVEL 1 VGROUPS 2 SINGLE_STABLE 0;
+sql CREATE DATABASE test1 VGROUPS 2;
 sql use test1;
 sql CREATE STABLE st (time TIMESTAMP, ca DOUBLE, cb DOUBLE, cc int) TAGS (ta VARCHAR(10) );
 
@@ -29,6 +30,76 @@ sql create stream streamd4 into streamt4 as select tbname, _wstart,_wend, count(
 sql create stream streamd5 into streamt5 as select tbname, _wstart,_wend, count(*), max(ca), min(cb) from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by tbname state_window(cc);
 sql create stream streamd6 into streamt6 as select ca, _wstart,_wend, count(*), max(ca), min(cb) from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca state_window(cc);
 
+
+sql alter local 'keepColumnName' '1'
+
+sql CREATE STABLE `meters_test_data` (`ts` TIMESTAMP, `close` FLOAT, `parttime` TIMESTAMP, `parttime_str` VARCHAR(32)) TAGS (`id` VARCHAR(32));
+
+sql_error create stream realtime_meters fill_history 1 into realtime_meters as select last(parttime),first(close),last(close) from meters_test_data partition by tbname state_window(parttime_str);
+sql_error create stream streamd7 into streamt7 as select  _wstart, _wend, count(*), first(ca), last(ca) from t1 interval(10s);
+sql_error create stream streamd71 into streamt71 as select _wstart, _wend, count(*) as ca, first(ca), last(ca) as c2 from t1 interval(10s);
+
+sql create stream streamd8 into streamt8 as select _wstart, _wend, count(*), first(ca) as c1, last(ca) as c2 from t1 interval(10s);
+sql desc streamt8;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql create stream streamd9 into streamt9 as select  _wstart, _wend, count(*), first(ca) as c1, last(ca) from t1 interval(10s);
+sql desc streamt9;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql_error create stream streamd11 into streamd11 as select _wstart, _wend, count(*), last(ca), last(ca) from t1 interval(10s);
+
+sql alter local 'keepColumnName' '0'
+
+sql create stream realtime_meters fill_history 1 into realtime_meters as select last(parttime),first(close),last(close) from meters_test_data partition by tbname state_window(parttime_str);
+
+sql desc realtime_meters;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql create stream streamd7 into streamt7 as select  _wstart, _wend, count(*), first(ca), last(ca) from t1 interval(10s);
+
+sql desc streamt7;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql create stream streamd71 into streamt71 as select _wstart, _wend, count(*) as ca, first(ca), last(ca) as c2 from t1 interval(10s);
+
+sql desc streamt71;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql create stream streamd10 into streamd10 as select _wstart, _wend, count(*), first(ca), last(cb) as c2 from t1 interval(10s);
+
+sql desc streamd10;
+
+if $rows == 0 then
+  return -1
+endi
+
+sql_error create stream streamd11 into streamd11 as select _wstart, _wend, count(*), last(ca), last(ca) from t1 interval(10s);
+
+
+sql create stream streamd12 into streamd12 as select _wstart, _wend, count(*), last(ca), last(cb) as c2 from t1 interval(10s);
+
+sql desc streamd12;
+
+if $rows == 0 then
+  return -1
+endi
+
 sleep 3000
 
 sql drop stream if exists streamd1;

From 0dd9e4bdeec44fbc18aa8530290bbfbf656097a0 Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Thu, 5 Jan 2023 11:35:39 +0800
Subject: [PATCH 11/13] fix(vnode): new vnodePostClose for closing sync module

---
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 7 +++++--
 source/dnode/vnode/inc/vnode.h           | 7 ++++---
 source/dnode/vnode/src/vnd/vnodeOpen.c   | 7 +++----
 3 files changed, 12 insertions(+), 9 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index 9e91670d33..6b274261e5 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -84,6 +84,9 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   taosThreadRwlockUnlock(&pMgmt->lock);
   vmReleaseVnode(pMgmt, pVnode);
 
+  dInfo("vgId:%d, pre close", pVnode->vgId);
+  vnodePreClose(pVnode->pImpl);
+
   dInfo("vgId:%d, wait for vnode ref become 0", pVnode->vgId);
   while (pVnode->refCount > 0) taosMsleep(10);
 
@@ -115,8 +118,8 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   dInfo("vgId:%d, all vnode queues is empty", pVnode->vgId);
 
-  dInfo("vgId:%d, pre close", pVnode->vgId);
-  vnodePreClose(pVnode->pImpl);
+  dInfo("vgId:%d, post close", pVnode->vgId);
+  vnodePostClose(pVnode->pImpl);
 
   vmFreeQueue(pMgmt, pVnode);
   vnodeClose(pVnode->pImpl);
diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h
index cbee70ad03..e92daaabe8 100644
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@@ -54,7 +54,8 @@ int32_t vnodeAlter(const char *path, SAlterVnodeReplicaReq *pReq, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
 SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
 void    vnodePreClose(SVnode *pVnode);
-void    vnodeSyncCheckTimeout(SVnode* pVnode);
+void    vnodePostClose(SVnode *pVnode);
+void    vnodeSyncCheckTimeout(SVnode *pVnode);
 void    vnodeClose(SVnode *pVnode);
 
 int32_t vnodeStart(SVnode *pVnode);
@@ -175,7 +176,7 @@ int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableL
 void         tsdbReaderClose(STsdbReader *pReader);
 bool         tsdbNextDataBlock(STsdbReader *pReader);
 void         tsdbRetrieveDataBlockInfo(const STsdbReader *pReader, int32_t *rows, uint64_t *uid, STimeWindow *pWindow);
-int32_t      tsdbRetrieveDatablockSMA(STsdbReader *pReader, SSDataBlock* pDataBlock, bool *allHave);
+int32_t      tsdbRetrieveDatablockSMA(STsdbReader *pReader, SSDataBlock *pDataBlock, bool *allHave);
 SSDataBlock *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
 int32_t      tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond);
 int32_t      tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
@@ -185,7 +186,7 @@ void        *tsdbGetIvtIdx(SMeta *pMeta);
 uint64_t     getReaderMaxVersion(STsdbReader *pReader);
 
 int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,
-                                uint64_t suid, void **pReader, const char* idstr);
+                                uint64_t suid, void **pReader, const char *idstr);
 int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray *pTableUids);
 void   *tsdbCacherowsReaderClose(void *pReader);
 int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid);
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index bec5d2977b..9affd534c7 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -245,10 +245,9 @@ _err:
   return NULL;
 }
 
-void vnodePreClose(SVnode *pVnode) {
-  vnodeQueryPreClose(pVnode);
-  vnodeSyncPreClose(pVnode);
-}
+void vnodePreClose(SVnode *pVnode) { vnodeQueryPreClose(pVnode); }
+
+void vnodePostClose(SVnode *pVnode) { vnodeSyncPreClose(pVnode); }
 
 void vnodeClose(SVnode *pVnode) {
   if (pVnode) {

From 7e5d97f0daa69fdef1805bd74d153fb4c99ea9a0 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Thu, 5 Jan 2023 12:00:27 +0800
Subject: [PATCH 12/13] fix: stop snapshot receiver on sync post close

---
 include/libs/sync/sync.h               |  1 +
 source/dnode/vnode/src/inc/vnd.h       |  1 +
 source/dnode/vnode/src/vnd/vnodeOpen.c |  7 +++++--
 source/dnode/vnode/src/vnd/vnodeSync.c |  5 +++++
 source/libs/sync/inc/syncInt.h         |  1 +
 source/libs/sync/src/syncMain.c        | 23 +++++++++++++++++++++++
 6 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h
index d37f8f76c2..f93bf9a326 100644
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@@ -232,6 +232,7 @@ int64_t syncOpen(SSyncInfo* pSyncInfo);
 int32_t syncStart(int64_t rid);
 void    syncStop(int64_t rid);
 void    syncPreStop(int64_t rid);
+void    syncPostStop(int64_t rid);
 int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak, int64_t* seq);
 int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg);
 int32_t syncReconfig(int64_t rid, SSyncCfg* pCfg);
diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h
index 28797c5361..93750ed585 100644
--- a/source/dnode/vnode/src/inc/vnd.h
+++ b/source/dnode/vnode/src/inc/vnd.h
@@ -98,6 +98,7 @@ bool    vnodeShouldRollback(SVnode* pVnode);
 int32_t vnodeSyncOpen(SVnode* pVnode, char* path);
 int32_t vnodeSyncStart(SVnode* pVnode);
 void    vnodeSyncPreClose(SVnode* pVnode);
+void    vnodeSyncPostClose(SVnode* pVnode);
 void    vnodeSyncClose(SVnode* pVnode);
 void    vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg, int32_t code);
 bool    vnodeIsLeader(SVnode* pVnode);
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 9affd534c7..cea85a0c11 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -245,9 +245,12 @@ _err:
   return NULL;
 }
 
-void vnodePreClose(SVnode *pVnode) { vnodeQueryPreClose(pVnode); }
+void vnodePreClose(SVnode *pVnode) {
+  vnodeQueryPreClose(pVnode);
+  vnodeSyncPreClose(pVnode);
+}
 
-void vnodePostClose(SVnode *pVnode) { vnodeSyncPreClose(pVnode); }
+void vnodePostClose(SVnode *pVnode) { vnodeSyncPostClose(pVnode); }
 
 void vnodeClose(SVnode *pVnode) {
   if (pVnode) {
diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c
index 749c81224c..65e17cfaad 100644
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
@@ -614,6 +614,11 @@ void vnodeSyncPreClose(SVnode *pVnode) {
   taosThreadMutexUnlock(&pVnode->lock);
 }
 
+void vnodeSyncPostClose(SVnode *pVnode) {
+  vInfo("vgId:%d, post close sync", pVnode->config.vgId);
+  syncPostStop(pVnode->sync);
+}
+
 void vnodeSyncClose(SVnode *pVnode) {
   vInfo("vgId:%d, close sync", pVnode->config.vgId);
   syncStop(pVnode->sync);
diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h
index 3bf4a8d1cd..6793430923 100644
--- a/source/libs/sync/inc/syncInt.h
+++ b/source/libs/sync/inc/syncInt.h
@@ -228,6 +228,7 @@ int32_t    syncNodeStart(SSyncNode* pSyncNode);
 int32_t    syncNodeStartStandBy(SSyncNode* pSyncNode);
 void       syncNodeClose(SSyncNode* pSyncNode);
 void       syncNodePreClose(SSyncNode* pSyncNode);
+void       syncNodePostClose(SSyncNode* pSyncNode);
 int32_t    syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak, int64_t *seq);
 int32_t    syncNodeRestore(SSyncNode* pSyncNode);
 void       syncHbTimerDataFree(SSyncHbTimerData* pData);
diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 32ae9e391c..5b9fb7c59c 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -124,6 +124,14 @@ void syncPreStop(int64_t rid) {
   }
 }
 
+void syncPostStop(int64_t rid) {
+  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+  if (pSyncNode != NULL) {
+    syncNodePostClose(pSyncNode);
+    syncNodeRelease(pSyncNode);
+  }
+}
+
 static bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pCfg) {
   if (!syncNodeInConfig(pSyncNode, pCfg)) return false;
   return abs(pCfg->replicaNum - pSyncNode->replicaNum) <= 1;
@@ -1236,6 +1244,7 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
     }
   }
 
+#if 0
   if (pSyncNode->pNewNodeReceiver != NULL) {
     if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
       snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
@@ -1246,6 +1255,7 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
     snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
     pSyncNode->pNewNodeReceiver = NULL;
   }
+#endif
 
   // stop elect timer
   syncNodeStopElectTimer(pSyncNode);
@@ -1257,6 +1267,19 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
   syncRespCleanRsp(pSyncNode->pSyncRespMgr);
 }
 
+void syncNodePostClose(SSyncNode* pSyncNode) {
+  if (pSyncNode->pNewNodeReceiver != NULL) {
+    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
+      snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
+    }
+
+    sDebug("vgId:%d, snapshot receiver destroy while preclose sync node, data:%p", pSyncNode->vgId,
+           pSyncNode->pNewNodeReceiver);
+    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
+    pSyncNode->pNewNodeReceiver = NULL;
+  }
+}
+
 void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); }
 
 void syncNodeClose(SSyncNode* pSyncNode) {

From ed56abd40f207dbdfd4e2cacb487f37de4b81ab9 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Thu, 5 Jan 2023 15:55:33 +0800
Subject: [PATCH 13/13] fix: crash if reader is already dropped

---
 source/libs/sync/src/syncSnapshot.c | 30 ++++++++++-------------------
 1 file changed, 10 insertions(+), 20 deletions(-)

diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c
index b68b735f46..defb7402f4 100644
--- a/source/libs/sync/src/syncSnapshot.c
+++ b/source/libs/sync/src/syncSnapshot.c
@@ -150,7 +150,7 @@ void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
 
 // when sender receive ack, call this function to send msg from seq
 // seq = ack + 1, already updated
-int32_t snapshotSend(SSyncSnapshotSender *pSender) {
+static int32_t snapshotSend(SSyncSnapshotSender *pSender) {
   // free memory last time (current seq - 1)
   if (pSender->pCurrentBlock != NULL) {
     taosMemoryFree(pSender->pCurrentBlock);
@@ -342,23 +342,6 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
 
 bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceiver->start; }
 
-// force stop
-void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
-  sRInfo(pReceiver, "snapshot receiver force stop, writer:%p");
-
-  // force close, abandon incomplete data
-  if (pReceiver->pWriter != NULL) {
-    int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false,
-                                                                  &pReceiver->snapshot);
-    if (ret != 0) {
-      sRInfo(pReceiver, "snapshot receiver force stop failed since %s", terrstr());
-    }
-    pReceiver->pWriter = NULL;
-  }
-
-  pReceiver->start = false;
-}
-
 static int32_t snapshotReceiverStartWriter(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) {
   if (pReceiver->pWriter != NULL) {
     sRError(pReceiver, "vgId:%d, snapshot receiver writer is not null");
@@ -590,7 +573,7 @@ _START_RECEIVER:
 
     if (snapshotReceiverIsStart(pReceiver)) {
       sRInfo(pReceiver, "snapshot receiver already start and force stop pre one");
-      snapshotReceiverForceStop(pReceiver);
+      snapshotReceiverStop(pReceiver);
     }
 
     snapshotReceiverStart(pReceiver, pMsg);  // set start-time same with sender
@@ -842,7 +825,7 @@ int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) {
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_FORCE_CLOSE) {
         // force close, no response
         syncLogRecvSyncSnapshotSend(pSyncNode, pMsg, "process force stop");
-        snapshotReceiverForceStop(pReceiver);
+        snapshotReceiverStop(pReceiver);
       } else if (pMsg->seq > SYNC_SNAPSHOT_SEQ_BEGIN && pMsg->seq < SYNC_SNAPSHOT_SEQ_END) {
         syncLogRecvSyncSnapshotSend(pSyncNode, pMsg, "process seq data");
         code = syncNodeOnSnapshotReceive(pSyncNode, pMsg);
@@ -989,6 +972,13 @@ int32_t syncNodeOnSnapshotRsp(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) {
     return syncNodeOnSnapshotPreRsp(pSyncNode, pSender, pMsg);
   }
 
+  if (pSender->pReader == NULL || pSender->finish) {
+    syncLogRecvSyncSnapshotRsp(pSyncNode, pMsg, "snapshot sender invalid");
+    sSError(pSender, "snapshot sender invalid, pReader:%p finish:%d", pMsg->code, pSender->pReader, pSender->finish);
+    terrno = pMsg->code;
+    goto _ERROR;
+  }
+
   if (pMsg->ack == SYNC_SNAPSHOT_SEQ_BEGIN) {
     syncLogRecvSyncSnapshotRsp(pSyncNode, pMsg, "process seq begin");
     if (snapshotSenderUpdateProgress(pSender, pMsg) != 0) {