Merge branch 'refact/submit_req_taosx' into refact/submit_req

include/common/tdataformat.h

@@ -146,6 +146,9 @@ extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max,
 int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind);
 void    tColDataSortMerge(SArray *colDataArr);
 
+//for raw block
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+                                    int32_t nRows, char* lengthOrbitmap, char *data);
 // for encode/decode
 int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
 int32_t tGetColData(uint8_t *pBuf, SColData *pColData);

include/libs/parser/parser.h

@@ -115,6 +115,8 @@ int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsS
                     char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash);
 
+int     rawBlockBindData(SQuery *query, STableMeta* pTableMeta, SRetrieveTableRsp* pRsp, SVCreateTbReq* pCreateTb);
+
 int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
 SArray* serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap);
 SArray* serializeVgroupsDropTableBatch(SHashObj* pVgroupHashmap);

source/common/src/tdataformat.c

@@ -2091,6 +2091,61 @@ _exit:
   return code;
 }
 
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+                                    int32_t nRows, char* lengthOrbitmap, char *data) {
+  int32_t code = 0;
+
+  if (IS_VAR_DATA_TYPE(type)) {  // var-length data type
+    for (int32_t i = 0; i < nRows; ++i) {
+      int32_t offset = *((int32_t*)lengthOrbitmap + i);
+      if (offset == -1) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+        if (code) goto _exit;
+      } else {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+            pColData, (uint8_t *)varDataVal(data + offset), varDataLen(data + offset));
+      }
+    }
+  } else {  // fixed-length data type
+    bool allValue = true;
+    bool allNull  = true;
+    for (int32_t i = 0; i < nRows; ++i) {
+      if(!colDataIsNull_f(lengthOrbitmap, i)){
+        allNull = false;
+      }else{
+        allValue = false;
+      }
+    }
+
+    if (allValue) {
+      // optimize (todo)
+      for (int32_t i = 0; i < nRows; ++i) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+            pColData, (uint8_t *)data + bytes * i, bytes);
+      }
+    } else if (allNull) {
+      // optimize (todo)
+      for (int32_t i = 0; i < nRows; ++i) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+        if (code) goto _exit;
+      }
+    } else {
+      for (int32_t i = 0; i < nRows; ++i) {
+        if (colDataIsNull_f(lengthOrbitmap, i)) {
+          code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+          if (code) goto _exit;
+        } else {
+          code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+              pColData, (uint8_t *)data + bytes * i, bytes);
+        }
+      }
+    }
+  }
+
+  _exit:
+  return code;
+}
+
 int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind) {
   int32_t code = 0;
 

source/libs/parser/src/parInsertUtil.c

@@ -20,6 +20,7 @@
 #include "parUtil.h"
 #include "querynodes.h"
 #include "tRealloc.h"
+#include "tdatablock.h"
 
 typedef struct SBlockKeyTuple {
   TSKEY   skey;
@@ -295,7 +296,7 @@ int32_t insGetDataBlockFromList(SHashObj* pHashList, void* id, int32_t idLen, in
   }
 
   if (*dataBlocks == NULL) {
-    int32_t ret = createDataBlock((size_t)size, rowSize, startOffset, pTableMeta, dataBlocks);
+    int32_t ret = createTableDataBlock((size_t)size, rowSize, startOffset, pTableMeta, dataBlocks);
     if (ret != TSDB_CODE_SUCCESS) {
       return ret;
     }
@@ -1371,3 +1372,77 @@ int32_t insBuildVgDataBlocks(SHashObj* pVgroupsHashObj, SArray* pVgDataCxtList,
 
   return code;
 }
+
+int rawBlockBindData(SQuery *query, STableMeta* pTableMeta, SRetrieveTableRsp* pRsp, SVCreateTbReq *pCreateTb){
+  STableDataCxt* pTableCxt = NULL;
+  int ret = insGetTableDataCxt(((SVnodeModifOpStmt *)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid, sizeof(pTableMeta->uid),
+                           pTableMeta, &pCreateTb, &pTableCxt, true);
+  if (ret != TSDB_CODE_SUCCESS) {
+    uError("insGetTableDataCxt error");
+    goto end;
+  }
+
+  // no need to bind, because select * get all fields
+  ret = initTableColSubmitData(pTableCxt);
+  if (ret != TSDB_CODE_SUCCESS) {
+    uError( "initTableColSubmitData error");
+    goto end;
+  }
+
+  char* p = (char*)pRsp->data;
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length |
+  p += sizeof(int32_t);
+  p += sizeof(int32_t);
+
+  int32_t numOfRows = *(int32_t*)p;
+  p += sizeof(int32_t);
+
+  int32_t numOfCols = *(int32_t*)p;
+  p += sizeof(int32_t);
+
+  p += sizeof(int32_t);
+  p += sizeof(uint64_t);
+
+  int8_t *fields = p;
+  p += numOfCols * (sizeof(int8_t) + sizeof(int32_t));
+
+  int32_t* colLength = (int32_t*)p;
+  p += sizeof(int32_t) * numOfCols;
+
+  char* pStart = p;
+
+  SSchema*            pSchema = getTableColumnSchema(pTableCxt->pMeta);
+  SBoundColInfo*      boundInfo = &pTableCxt->boundColsInfo;
+
+  if(boundInfo->numOfBound != numOfCols){
+    uError("boundInfo->numOfBound:%d != numOfCols:%d", boundInfo->numOfBound, numOfCols);
+    ret = TSDB_CODE_INVALID_PARA;
+    goto end;
+  }
+  for (int c = 0; c < boundInfo->numOfBound; ++c) {
+    SSchema* pColSchema = &pSchema[boundInfo->pColIndex[c]];
+    SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c);
+
+    if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
+      uError( "type or bytes not equal");
+      ret = TSDB_CODE_INVALID_PARA;
+      goto end;
+    }
+
+    colLength[c] = htonl(colLength[c]);
+    int8_t* offset = pStart;
+    if (IS_VAR_DATA_TYPE(pColSchema->type)) {
+      pStart += numOfRows * sizeof(int32_t);
+    } else {
+      pStart += BitmapLen(numOfRows);
+    }
+    char *pData = pStart;
+
+    tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData);
+    fields += sizeof(int8_t) + sizeof(int32_t);
+    pStart += colLength[c];
+  }
+
+end:
+  return ret;
+}
\ No newline at end of file