Merge branch 'fix/TD-21746' into FIX/TD-21043-main

include/libs/sync/sync.h

@@ -234,6 +234,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);

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -120,6 +120,9 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   dInfo("vgId:%d, all vnode queues is empty", pVnode->vgId);
 
+  dInfo("vgId:%d, post close", pVnode->vgId);
+  vnodePostClose(pVnode->pImpl);
+
   vmFreeQueue(pMgmt, pVnode);
   vnodeClose(pVnode->pImpl);
   pVnode->pImpl = NULL;

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);

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);
@@ -176,7 +177,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);
@@ -186,7 +187,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);

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

@@ -99,6 +99,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);

source/dnode/vnode/src/vnd/vnodeOpen.c

@@ -252,6 +252,8 @@ void vnodePreClose(SVnode *pVnode) {
   vnodeSyncPreClose(pVnode);
 }
 
+void vnodePostClose(SVnode *pVnode) { vnodeSyncPostClose(pVnode); }
+
 void vnodeClose(SVnode *pVnode) {
   if (pVnode) {
     tsem_wait(&pVnode->canCommit);

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -648,6 +648,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);

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);
 

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;
@@ -745,7 +745,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 {
@@ -759,7 +762,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;
+              }
             }
           }
         }
@@ -773,7 +779,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;
+              }
             }
           }
         }
@@ -787,7 +796,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;
+              }
             }
           }
         }
@@ -803,14 +815,17 @@ 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;
+            }
           }
         }
       }
     }
 
     pBuf->assign = true;
-    return numOfElems;
+    return TSDB_CODE_SUCCESS;
   }
 
   int32_t start = pInput->startRowIndex;
@@ -824,7 +839,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;
@@ -887,9 +905,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;
 }

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;
@@ -380,6 +386,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);
     }
@@ -391,7 +400,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
   }
 
   pBucket->total += count;
-  return 0;
+  return TSDB_CODE_SUCCESS;
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////
@@ -471,6 +480,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;
@@ -505,8 +517,14 @@ 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);
-
-          tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
+          if (pg == NULL) {
+            return -1;
+          }
+
+          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);
         }
@@ -528,7 +546,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);

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);

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;
@@ -1238,6 +1246,7 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
     }
   }
 
+#if 0
   if (pSyncNode->pNewNodeReceiver != NULL) {
     if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
       snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
@@ -1248,6 +1257,7 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
     snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
     pSyncNode->pNewNodeReceiver = NULL;
   }
+#endif
 
   // stop elect timer
   syncNodeStopElectTimer(pSyncNode);
@@ -1259,6 +1269,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) {

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) {

tests/script/tsim/stream/basic3.sim

 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;