fix(stream): msg deserialize

source/common/src/tdatablock.c

@@ -99,7 +99,7 @@ void colDataTrim(SColumnInfoData* pColumnInfoData) {
   // TODO
 }
 
-int32_t getJsonValueLen(const char *data) {
+int32_t getJsonValueLen(const char* data) {
   int32_t dataLen = 0;
   if (*data == TSDB_DATA_TYPE_NULL) {
     dataLen = CHAR_BYTES;
@@ -109,7 +109,7 @@ int32_t getJsonValueLen(const char *data) {
     dataLen = DOUBLE_BYTES + CHAR_BYTES;
   } else if (*data == TSDB_DATA_TYPE_BOOL) {
     dataLen = CHAR_BYTES + CHAR_BYTES;
-  } else if (*data & TD_TAG_JSON) {   // json string
+  } else if (*data & TD_TAG_JSON) {  // json string
     dataLen = ((STag*)(data))->len;
   } else {
     ASSERT(0);
@@ -137,7 +137,7 @@ int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, con
     int32_t dataLen = 0;
     if (type == TSDB_DATA_TYPE_JSON) {
       dataLen = getJsonValueLen(pData);
-    }else {
+    } else {
       dataLen = varDataTLen(pData);
     }
 
@@ -1283,7 +1283,7 @@ static void doShiftBitmap(char* nullBitmap, size_t n, size_t total) {
   if (n % 8 == 0) {
     memmove(nullBitmap, nullBitmap + n / 8, newLen);
   } else {
-    int32_t tail = n % 8;
+    int32_t  tail = n % 8;
     int32_t  i = 0;
     uint8_t* p = (uint8_t*)nullBitmap;
 
@@ -1301,7 +1301,7 @@ static void doShiftBitmap(char* nullBitmap, size_t n, size_t total) {
       }
     } else if (n > 8) {
       int32_t gap = len - newLen;
-      while(i < newLen) {
+      while (i < newLen) {
         uint8_t v = p[i + gap];
         p[i] = (v << tail);
 
@@ -1316,7 +1316,6 @@ static void doShiftBitmap(char* nullBitmap, size_t n, size_t total) {
   }
 }
 
-
 static void colDataTrimFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) {
   if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
     memmove(pColInfoData->varmeta.offset, &pColInfoData->varmeta.offset[n], (total - n) * sizeof(int32_t));
@@ -1544,7 +1543,8 @@ void blockDebugShowData(const SArray* dataBlocks, const char* flag) {
  *
  * TODO: colId should be set
  */
-int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId, tb_uid_t suid) {
+int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
+                                    tb_uid_t suid) {
   int32_t sz = taosArrayGetSize(pDataBlocks);
   int32_t bufSize = sizeof(SSubmitReq);
   for (int32_t i = 0; i < sz; ++i) {
@@ -1585,11 +1585,11 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
     int32_t dataLen = 0;
     for (int32_t j = 0; j < rows; ++j) {                     // iterate by row
       tdSRowResetBuf(&rb, POINTER_SHIFT(pDataBuf, msgLen));  // set row buf
-      bool isStartKey = false;
+      bool    isStartKey = false;
       int32_t offset = 0;
       for (int32_t k = 0; k < colNum; ++k) {  // iterate by column
         SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
-        STColumn*         pCol = &pTSchema->columns[k];
+        STColumn*        pCol = &pTSchema->columns[k];
         ASSERT(pCol->type == pColInfoData->info.type);
         void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
         switch (pColInfoData->info.type) {
@@ -1600,15 +1600,18 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
                                   offset, k);
 
             } else {
-              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true, offset, k);
+              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var,
+                                  true, offset, k);
             }
             break;
           case TSDB_DATA_TYPE_NCHAR: {
-            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_NCHAR, TD_VTYPE_NORM, var, true, offset, k);
+            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_NCHAR, TD_VTYPE_NORM, var, true,
+                                offset, k);
             break;
           }
           case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
-            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_VARCHAR, TD_VTYPE_NORM, var, true, offset, k);
+            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_VARCHAR, TD_VTYPE_NORM, var, true,
+                                offset, k);
             break;
           }
           case TSDB_DATA_TYPE_VARBINARY:
@@ -1620,7 +1623,8 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
             break;
           default:
             if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
-              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pColInfoData->info.type, TD_VTYPE_NORM, var, true, offset, k);
+              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pColInfoData->info.type, TD_VTYPE_NORM, var,
+                                  true, offset, k);
             } else {
               uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
               TASSERT(0);
@@ -1667,7 +1671,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
                             const char* stbFullName, int32_t vgId) {
   SSubmitReq* ret = NULL;
   SArray*     tagArray = taosArrayInit(1, sizeof(STagVal));
-  if(!tagArray) {
+  if (!tagArray) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
@@ -1692,8 +1696,6 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
       createTbReq.type = TSDB_CHILD_TABLE;
       createTbReq.ctb.suid = suid;
 
-
-
       STagVal tagVal = {.cid = 1,
                         .type = TSDB_DATA_TYPE_UBIGINT,
                         .pData = (uint8_t*)&pDataBlock->info.groupId,
@@ -1831,11 +1833,12 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
   }
 
   ret->length = htonl(ret->length);
-        taosArrayDestroy(tagArray);
+  taosArrayDestroy(tagArray);
   return ret;
 }
 
-void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols, int8_t needCompress) {
+void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
+                         int8_t needCompress) {
   int32_t* actualLen = (int32_t*)data;
   data += sizeof(int32_t);
 
@@ -1929,4 +1932,4 @@ const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t
 
   ASSERT(pStart - pData == dataLen);
   return pStart;
-}
\ No newline at end of file
+}

source/libs/stream/inc/streamInc.h

@@ -27,6 +27,8 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamSink1(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDataBlock* data);
 
+int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/stream/src/stream.c

@@ -35,18 +35,19 @@ int32_t streamTriggerByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb) {
   return 0;
 }
 
-#if 1
 int32_t streamTaskEnqueue(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pRsp) {
-  SStreamDataBlock* pBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
+  SStreamDataBlock* pData = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
   int8_t            status;
 
   // enqueue
-  if (pBlock != NULL) {
-    pBlock->type = STREAM_DATA_TYPE_SSDATA_BLOCK;
-    pBlock->sourceVg = pReq->sourceVg;
-    pBlock->blocks = pReq->data;
+  if (pData != NULL) {
+    pData->type = STREAM_DATA_TYPE_SSDATA_BLOCK;
+    pData->sourceVg = pReq->sourceVg;
+    // decode
+    /*pData->blocks = pReq->data;*/
     /*pBlock->sourceVer = pReq->sourceVer;*/
-    if (streamTaskInput(pTask, (SStreamQueueItem*)pBlock) == 0) {
+    streamDispatchReqToData(pReq, pData);
+    if (streamTaskInput(pTask, (SStreamQueueItem*)pData) == 0) {
       status = TASK_INPUT_STATUS__NORMAL;
     } else {
       status = TASK_INPUT_STATUS__FAILED;
@@ -68,7 +69,6 @@ int32_t streamTaskEnqueue(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg*
   tmsgSendRsp(pRsp);
   return status == TASK_INPUT_STATUS__NORMAL ? 0 : -1;
 }
-#endif
 
 int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pRsp) {
   // 1. handle input

source/libs/stream/src/streamData.c

@@ -36,6 +36,29 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput) {
 }
 #endif
 
+int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData) {
+  int32_t blockNum = pReq->blockNum;
+  SArray* pArray = taosArrayInit(blockNum, sizeof(SSDataBlock));
+  if (pArray == NULL) {
+    return -1;
+  }
+  taosArraySetSize(pArray, blockNum);
+
+  ASSERT(pReq->blockNum == taosArrayGetSize(pReq->data));
+  ASSERT(pReq->blockNum == taosArrayGetSize(pReq->dataLen));
+
+  for (int32_t i = 0; i < blockNum; i++) {
+    int32_t            len = *(int32_t*)taosArrayGet(pReq->dataLen, i);
+    SRetrieveTableRsp* pRetrieve = taosArrayGetP(pReq->data, i);
+    SSDataBlock*       pDataBlock = taosArrayGet(pArray, i);
+    blockCompressDecode(pDataBlock, htonl(pRetrieve->numOfCols), htonl(pRetrieve->numOfRows), pRetrieve->data);
+    // TODO: refactor
+    pDataBlock->info.childId = pReq->sourceChildId;
+  }
+  pData->blocks = pArray;
+  return 0;
+}
+
 SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq) {
   SStreamDataSubmit* pDataSubmit = (SStreamDataSubmit*)taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
   if (pDataSubmit == NULL) return NULL;

source/libs/stream/src/streamMsg.c

@@ -71,6 +71,7 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis
   pRetrieve->compressed = 0;
   pRetrieve->completed = 1;
   pRetrieve->numOfRows = htonl(pBlock->info.rows);
+  pRetrieve->numOfCols = htonl(pBlock->info.numOfCols);
 
   int32_t actualLen = 0;
   blockCompressEncode(pBlock, pRetrieve->data, &actualLen, pBlock->info.numOfCols, false);