diff --git a/examples/c/stream_demo.c b/examples/c/stream_demo.c
index 961eb6c93ad51ba785a9075c2e026fa20f7a8446..a4fc30ff65d63d435eb533f1f2624e0694e1d6b0 100644
--- a/examples/c/stream_demo.c
+++ b/examples/c/stream_demo.c
@@ -25,7 +25,7 @@ int32_t init_env() {
     return -1;
   }
 
-  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
+  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
   if (taos_errno(pRes) != 0) {
     printf("error in create db, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -68,6 +68,14 @@ int32_t init_env() {
     return -1;
   }
   taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table if not exists tu3 using st1 tags(3)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
   return 0;
 }
 
@@ -90,10 +98,9 @@ int32_t create_stream() {
   /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
-  pRes = taos_query(
-      pConn,
-      "create stream stream1 trigger window_close watermark 10s into outstb as select _wstartts, sum(k) from st1 "
-      "interval(10s) ");
+  pRes = taos_query(pConn,
+                    "create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from st1 "
+                    "partition by tbname interval(10s) ");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index 0243f2a9f01ed83256d07c850d485638b979bbc9..6184d8f81092d491fd6defd8aa93c884722a2cdc 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -243,7 +243,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle, uint64_
       if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
         break;
       }
-      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
+      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
         goto FAIL;
       }
     }
diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index 179dc88d2a27a8aaba95c268d12fec5329dd7f08..a8b18210dd51f0baff3b649e372adeb8d4783ac0 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -63,7 +63,6 @@ int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) {
 }
 
 int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq) {
-  //
   if (tStartEncode(pEncoder) < 0) return -1;
   if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1;
   if (tEncodeI32(pEncoder, pReq->dstNodeId) < 0) return -1;
@@ -84,7 +83,7 @@ int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) {
   if (tDecodeI32(pDecoder, &pReq->srcTaskId) < 0) return -1;
   uint64_t len = 0;
   if (tDecodeBinaryAlloc(pDecoder, (void**)&pReq->pRetrieve, &len) < 0) return -1;
-  pReq->retrieveLen = len;
+  pReq->retrieveLen = (int32_t)len;
   tEndDecode(pDecoder);
   return 0;
 }