util->cmakefile.txt conflict

documentation20/cn/02.getting-started/docs.md

@@ -126,7 +126,7 @@ taos> source <filename>;
 $ taosdemo
 ```
 
-该命令将在数据库 test 下面自动创建一张超级表 meters，该超级表下有 1 万张表，表名为 "t0" 到 "t9999"，每张表有 1 万条记录，每条记录有 (ts, current, voltage, phase) 四个字段，时间戳从 "2017-07-14 10:40:00 000" 到 "2017-07-14 10:40:09 999"，每张表带有标签 location 和 groupdId，groupdId 被设置为 1 到 10， location 被设置为 "beijing" 或者 "shanghai"。
+该命令将在数据库 test 下面自动创建一张超级表 meters，该超级表下有 1 万张表，表名为 "d0" 到 "d9999"，每张表有 1 万条记录，每条记录有 (ts, current, voltage, phase) 四个字段，时间戳从 "2017-07-14 10:40:00 000" 到 "2017-07-14 10:40:09 999"，每张表带有标签 location 和 groupdId，groupdId 被设置为 1 到 10， location 被设置为 "beijing" 或者 "shanghai"。
 
 执行这条命令大概需要几分钟，最后共插入 1 亿条记录。
 
@@ -156,10 +156,10 @@ taos> select count(*) from test.meters where location="beijing";
 taos> select avg(current), max(voltage), min(phase) from test.meters where groupdId=10;
 ```
 
-- 对表 t10 按 10s 进行平均值、最大值和最小值聚合统计：
+- 对表 d10 按 10s 进行平均值、最大值和最小值聚合统计：
 
 ```mysql
-taos> select avg(current), max(voltage), min(phase) from test.t10 interval(10s);
+taos> select avg(current), max(voltage), min(phase) from test.d10 interval(10s);
 ```
 
 **Note:** taosdemo 命令本身带有很多选项，配置表的数目、记录条数等等，请执行 `taosdemo --help` 详细列出。您可以设置不同参数进行体验。

src/client/inc/tscUtil.h

@@ -94,11 +94,22 @@ typedef struct SVgroupTableInfo {
   SArray     *itemList;   // SArray<STableIdInfo>
 } SVgroupTableInfo;
 
+typedef struct SBlockKeyTuple {
+  TSKEY skey;
+  void* payloadAddr;
+} SBlockKeyTuple;
+
+typedef struct SBlockKeyInfo {
+  int32_t         maxBytesAlloc;
+  SBlockKeyTuple* pKeyTuple;
+} SBlockKeyInfo;
+
 int32_t converToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len);
 
 int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, SName* name, STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
 void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta);
-void tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf);
+void    tscSortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf);
+int     tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKeyInfo);
 
 void tscDestroyBoundColumnInfo(SParsedDataColInfo* pColInfo);
 void doRetrieveSubqueryData(SSchedMsg *pMsg);

src/client/inc/tsclient.h

@@ -88,13 +88,43 @@ typedef struct SBoundColumn {
   int32_t offset;  // all column offset value
 } SBoundColumn;
 
+typedef struct {
+  uint16_t schemaColIdx;
+  uint16_t boundIdx;
+  uint16_t finalIdx;
+} SBoundIdxInfo;
+
+typedef enum _COL_ORDER_STATUS {
+  ORDER_STATUS_UNKNOWN = 0,
+  ORDER_STATUS_ORDERED = 1,
+  ORDER_STATUS_DISORDERED = 2,
+} EOrderStatus;
+
 typedef struct SParsedDataColInfo {
-  int16_t         numOfCols;
-  int16_t         numOfBound;
-  int32_t        *boundedColumns;
-  SBoundColumn   *cols;
+  int16_t        numOfCols;
+  int16_t        numOfBound;
+  int32_t *      boundedColumns;  // bounded column idx according to schema
+  SBoundColumn * cols;
+  SBoundIdxInfo *colIdxInfo;
+  int8_t         orderStatus;  // bounded columns: 
 } SParsedDataColInfo;
 
+#define IS_DATA_COL_ORDERED(s) ((s) == (int8_t)ORDER_STATUS_ORDERED)
+
+typedef struct {
+  SSchema *   pSchema;
+  int16_t     sversion;
+  int32_t     flen;
+  uint16_t    nCols;
+  void *      buf;
+  void *      pDataBlock;
+  SSubmitBlk *pSubmitBlk;
+} SMemRowBuilder;
+
+typedef struct {
+  TDRowLenT allNullLen;
+} SMemRowHelper;
+
 typedef struct STableDataBlocks {
   SName       tableName;
   int8_t      tsSource;     // where does the UNIX timestamp come from, server or client
@@ -109,12 +139,13 @@ typedef struct STableDataBlocks {
   STableMeta *pTableMeta;   // the tableMeta of current table, the table meta will be used during submit, keep a ref to avoid to be removed from cache
   char       *pData;
 
-  SParsedDataColInfo  boundColumnInfo;
+  SParsedDataColInfo boundColumnInfo;
 
   // for parameter ('?') binding
-  uint32_t    numOfAllocedParams;
-  uint32_t    numOfParams;
-  SParamInfo *params;
+  uint32_t       numOfAllocedParams;
+  uint32_t       numOfParams;
+  SParamInfo *   params;
+  SMemRowHelper  rowHelper;
 } STableDataBlocks;
 
 typedef struct {
@@ -128,6 +159,7 @@ typedef struct SInsertStatementParam {
   SHashObj    *pTableBlockHashList;     // data block for each table
   SArray      *pDataBlocks;             // SArray<STableDataBlocks*>. Merged submit block for each vgroup
   int8_t       schemaAttached;          // denote if submit block is built with table schema or not
+  uint8_t      payloadType;             // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
   STagData     tagData;                 // NOTE: pTagData->data is used as a variant length array
 
   int32_t      batchSize;               // for parameter ('?') binding and batch processing
@@ -139,6 +171,14 @@ typedef struct SInsertStatementParam {
   char        *sql;                     // current sql statement position
 } SInsertStatementParam;
 
+typedef enum {
+  PAYLOAD_TYPE_KV = 0,
+  PAYLOAD_TYPE_RAW = 1,
+} EPayloadType;
+
+#define IS_RAW_PAYLOAD(t) \
+  (((int)(t)) == PAYLOAD_TYPE_RAW)  // 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
+
 // TODO extract sql parser supporter
 typedef struct {
   int     command;
@@ -243,6 +283,7 @@ typedef struct SSqlObj {
   void *           pStream;
   void *           pSubscription;
   char *           sqlstr;
+  void *           pBuf;  // table meta buffer
   char             parseRetry;
   char             retry;
   char             maxRetry;
@@ -382,12 +423,17 @@ extern int   tscRefId;
 extern int   tscNumOfObj;     // number of existed sqlObj in current process.
 
 extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
-
+ 
 void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
 int16_t getNewResColId(SSqlCmd* pCmd);
 
+int32_t schemaIdxCompar(const void *lhs, const void *rhs);
+int32_t boundIdxCompar(const void *lhs, const void *rhs);
+int     initSMemRowHelper(SMemRowHelper *pHelper, SSchema *pSSchema, uint16_t nCols, uint16_t allNullColsLen);
+int32_t getExtendedRowSize(STableComInfo *tinfo);
+
 #ifdef __cplusplus
 }
 #endif
 
-#endif
+#endif
\ No newline at end of file

src/client/src/taos.def

@@ -7,11 +7,16 @@ taos_connect_auth
 taos_close
 taos_stmt_init
 taos_stmt_prepare
+taos_stmt_set_tbname_tags
+taos_stmt_set_tbname
+taos_stmt_is_insert
+taos_stmt_num_params
 taos_stmt_bind_param
 taos_stmt_add_batch
 taos_stmt_execute
 taos_stmt_use_result
 taos_stmt_close
+taos_stmt_errstr
 taos_query
 taos_fetch_row
 taos_result_precision
@@ -37,6 +42,4 @@ taos_consume
 taos_unsubscribe
 taos_open_stream
 taos_close_stream
-taos_fetch_block
 taos_load_table_info
-

src/client/src/tscPrepare.c

@@ -1806,6 +1806,7 @@ int taos_stmt_execute(TAOS_STMT* stmt) {
 
     pStmt->last = STMT_EXECUTE;
 
+    pStmt->pSql->cmd.insertParam.payloadType = PAYLOAD_TYPE_RAW;
     if (pStmt->multiTbInsert) {
       ret = insertBatchStmtExecute(pStmt);
     } else {

src/client/src/tscSQLParser.c

@@ -1396,12 +1396,16 @@ static bool validateTableColumnInfo(SArray* pFieldList, SSqlCmd* pCmd) {
   const char* msg4 = "invalid data type";
   const char* msg5 = "invalid binary/nchar column length";
   const char* msg6 = "invalid column name";
+  const char* msg7 = "too many columns";
 
   // number of fields no less than 2
   size_t numOfCols = taosArrayGetSize(pFieldList);
-  if (numOfCols <= 1 || numOfCols > TSDB_MAX_COLUMNS) {
+  if (numOfCols <= 1 ) {
     invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg);
     return false;
+  } else if (numOfCols > TSDB_MAX_COLUMNS) {
+    invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7);
+    return false;
   }
 
   // first column must be timestamp
@@ -1536,13 +1540,20 @@ bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField) {
   const char* msg4 = "invalid tag name";
   const char* msg5 = "invalid binary/nchar tag length";
   const char* msg6 = "invalid data type in tags";
+  const char* msg7 = "too many columns";
 
   STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd,  0);
   STableMeta*     pTableMeta = pTableMetaInfo->pTableMeta;
 
   int32_t numOfTags = tscGetNumOfTags(pTableMeta);
   int32_t numOfCols = tscGetNumOfColumns(pTableMeta);
-  
+
+  // no more max columns
+  if (numOfTags + numOfCols >= TSDB_MAX_COLUMNS) {
+    invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7);
+    return false;
+  }
+
   // no more than 6 tags
   if (numOfTags == TSDB_MAX_TAGS) {
     char msg[128] = {0};
@@ -5999,6 +6010,16 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       return invalidOperationMsg(pMsg, msg22);
     }
 
+    SSchema* pSchema = (SSchema*) pTableMetaInfo->pTableMeta->schema;
+    int16_t numOfColumns = pTableMetaInfo->pTableMeta->tableInfo.numOfColumns;
+    int16_t i;
+    uint32_t nLen = 0;
+    for (i = 0; i < numOfColumns; ++i) {
+      nLen += pSchema[i].colId != columnIndex.columnIndex ? pSchema[i].bytes : pItem->bytes;
+    }
+    if (nLen >= TSDB_MAX_BYTES_PER_ROW) {
+      return invalidOperationMsg(pMsg, msg24);
+    }
     TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
     tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
   }else if (pAlterSQL->type == TSDB_ALTER_TABLE_MODIFY_TAG_COLUMN) {
@@ -6040,6 +6061,17 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       return invalidOperationMsg(pMsg, msg22);
     }
 
+    SSchema* pSchema = (SSchema*) pTableMetaInfo->pTableMeta->schema;
+    int16_t numOfColumns = pTableMetaInfo->pTableMeta->tableInfo.numOfColumns;
+    int16_t i;
+    uint32_t nLen = 0;
+    for (i = 0; i < numOfColumns; ++i) {
+      nLen += pSchema[i].colId != columnIndex.columnIndex ? pSchema[i].bytes : pItem->bytes;
+    }
+    if (nLen >= TSDB_MAX_BYTES_PER_ROW) {
+      return invalidOperationMsg(pMsg, msg24);
+    }
+
     TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
     tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
   }
@@ -8005,6 +8037,28 @@ static void freeElem(void* p) {
   tfree(*(char**)p);
 }
 
+int32_t tnameComparFn(const void* p1, const void* p2) {
+  SName* pn1 = (SName*)p1;
+  SName* pn2 = (SName*)p2;
+
+  int32_t ret = strncmp(pn1->acctId, pn2->acctId, tListLen(pn1->acctId));
+  if (ret != 0) {
+    return ret > 0? 1:-1;
+  } else {
+    ret = strncmp(pn1->dbname, pn2->dbname, tListLen(pn1->dbname));
+    if (ret != 0) {
+      return ret > 0? 1:-1;
+    } else {
+      ret = strncmp(pn1->tname, pn2->tname, tListLen(pn1->tname));
+      if (ret != 0) {
+        return ret > 0? 1:-1;
+      } else {
+        return 0;
+      }
+    }
+  }
+}
+
 int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   SSqlCmd* pCmd = &pSql->cmd;
 
@@ -8048,13 +8102,20 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   uint32_t maxSize = tscGetTableMetaMaxSize();
   char     name[TSDB_TABLE_FNAME_LEN] = {0};
 
-  char buf[80 * 1024] = {0};
-  assert(maxSize < 80 * 1024);
+  assert(maxSize < 80 * TSDB_MAX_COLUMNS);
+  if (!pSql->pBuf) {
+    if (NULL == (pSql->pBuf = tcalloc(1, 80 * TSDB_MAX_COLUMNS))) {
+      return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    }
+  }
   pTableMeta = calloc(1, maxSize);
 
   plist = taosArrayInit(4, POINTER_BYTES);
   pVgroupList = taosArrayInit(4, POINTER_BYTES);
 
+  taosArraySort(tableNameList, tnameComparFn);
+  taosArrayRemoveDuplicate(tableNameList, tnameComparFn, NULL);
+
   size_t numOfTables = taosArrayGetSize(tableNameList);
   for (int32_t i = 0; i < numOfTables; ++i) {
     SName* pname = taosArrayGet(tableNameList, i);
@@ -8066,7 +8127,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
 
     if (pTableMeta->id.uid > 0) {
       if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
-        code = tscCreateTableMetaFromSTableMeta(pTableMeta, name, buf);
+        code = tscCreateTableMetaFromSTableMeta(pTableMeta, name, pSql->pBuf);
 
         // create the child table meta from super table failed, try load it from mnode
         if (code != TSDB_CODE_SUCCESS) {
@@ -8075,8 +8136,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
           continue;
         }
       } else if (pTableMeta->tableType == TSDB_SUPER_TABLE) {
-        // the vgroup list of a super table is not kept in local buffer, so here need retrieve it
-        // from the mnode each time
+        // the vgroup list of super table is not kept in local buffer, so here need retrieve it from the mnode each time
         char* t = strdup(name);
         taosArrayPush(pVgroupList, &t);
       }
@@ -8103,6 +8163,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   if (pInfo->funcs) {
     funcSize = taosArrayGetSize(pInfo->funcs);
   }
+
   if (funcSize > 0) {
     for (size_t i = 0; i < funcSize; ++i) {
       SStrToken* t = taosArrayGet(pInfo->funcs, i);
@@ -8262,7 +8323,9 @@ static int32_t doValidateSubquery(SSqlNode* pSqlNode, int32_t index, SSqlObj* pS
 
   // union all is not support currently
   SSqlNode* p = taosArrayGetP(subInfo->pSubquery, 0);
-
+  if (taosArrayGetSize(subInfo->pSubquery) >= 2) {
+    return invalidOperationMsg(msgBuf, "not support union in subquery");
+  }
   SQueryInfo* pSub = calloc(1, sizeof(SQueryInfo));
   tscInitQueryInfo(pSub);
 

src/client/src/tscServer.c

@@ -2804,22 +2804,24 @@ int32_t tscGetTableMetaImpl(SSqlObj* pSql, STableMetaInfo *pTableMetaInfo, bool
   taosHashGetClone(tscTableMetaInfo, name, len, NULL, pTableMetaInfo->pTableMeta, -1);
 
   // TODO resize the tableMeta
-  char buf[80*1024] = {0};
-  assert(size < 80*1024);
+  assert(size < 80 * TSDB_MAX_COLUMNS);
+  if (!pSql->pBuf) {
+    if (NULL == (pSql->pBuf = tcalloc(1, 80 * TSDB_MAX_COLUMNS))) {
+      return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    }
+  }
 
   STableMeta* pMeta = pTableMetaInfo->pTableMeta;
   if (pMeta->id.uid > 0) {
     // in case of child table, here only get the
     if (pMeta->tableType == TSDB_CHILD_TABLE) {
-      int32_t code = tscCreateTableMetaFromSTableMeta(pTableMetaInfo->pTableMeta, name, buf);
+      int32_t code = tscCreateTableMetaFromSTableMeta(pTableMetaInfo->pTableMeta, name, pSql->pBuf);
       if (code != TSDB_CODE_SUCCESS) {
         return getTableMetaFromMnode(pSql, pTableMetaInfo, autocreate);
       }
     }
-
     return TSDB_CODE_SUCCESS;
   }
-
   return getTableMetaFromMnode(pSql, pTableMetaInfo, autocreate);
 }
 
@@ -2934,7 +2936,6 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, SQueryInfo* pQueryInfo) {
   if (allVgroupInfoRetrieved(pQueryInfo)) {
     return TSDB_CODE_SUCCESS;
   }
-
   SSqlObj *pNew = calloc(1, sizeof(SSqlObj));
   pNew->pTscObj = pSql->pTscObj;
   pNew->signature = pNew;

src/client/src/tscSubquery.c

@@ -2436,9 +2436,9 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
   tOrderDescriptor *pDesc  = NULL;
 
   pRes->qId = 0x1;  // hack the qhandle check
-  
-  const uint32_t nBufferSize = (1u << 16u);  // 64KB
-  
+
+  const uint32_t nBufferSize = (1u << 18u);  // 256KB
+
   SQueryInfo     *pQueryInfo = tscGetQueryInfo(pCmd);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
   SSubqueryState *pState = &pSql->subState;

src/client/src/tscUtil.c

@@ -1315,8 +1315,7 @@ void tscFreeQueryInfo(SSqlCmd* pCmd, bool removeMeta) {
     return;
   }
 
-  SQueryInfo* pQueryInfo = tscGetQueryInfo(pCmd);
-
+  SQueryInfo* pQueryInfo = pCmd->pQueryInfo;
   while(pQueryInfo != NULL) {
     SQueryInfo* p = pQueryInfo->sibling;
 
@@ -1488,6 +1487,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
   pSql->signature = NULL;
   pSql->fp = NULL;
   tfree(pSql->sqlstr);
+  tfree(pSql->pBuf);
 
   tfree(pSql->pSubs);
   pSql->subState.numOfSub = 0;
@@ -1499,7 +1499,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
   memset(pCmd->payload, 0, (size_t)pCmd->allocSize);
   tfree(pCmd->payload);
   pCmd->allocSize = 0;
-  
+
   tsem_destroy(&pSql->rspSem);
   memset(pSql, 0, sizeof(*pSql));
   free(pSql);
@@ -1508,6 +1508,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
 void tscDestroyBoundColumnInfo(SParsedDataColInfo* pColInfo) {
   tfree(pColInfo->boundedColumns);
   tfree(pColInfo->cols);
+  tfree(pColInfo->colIdxInfo);
 }
 
 void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta) {
@@ -1596,7 +1597,7 @@ void freeUdfInfo(SUdfInfo* pUdfInfo) {
   taosCloseDll(pUdfInfo->handle);
 }
 
-
+// todo refactor
 void*  tscDestroyUdfArrayList(SArray* pUdfList) {
   if (pUdfList == NULL) {
     return NULL;
@@ -1762,11 +1763,108 @@ int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, i
   return TSDB_CODE_SUCCESS;
 }
 
-static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bool includeSchema) {
+static SMemRow tdGenMemRowFromBuilder(SMemRowBuilder* pBuilder) {
+  SSchema* pSchema = pBuilder->pSchema;
+  char*    p = (char*)pBuilder->buf;
+  int      toffset = 0;
+  uint16_t nCols = pBuilder->nCols;
+
+  uint8_t  memRowType = payloadType(p);
+  uint16_t nColsBound = payloadNCols(p);
+  if (pBuilder->nCols <= 0 || nColsBound <= 0) {
+    return NULL;
+  }
+  char*    pVals = POINTER_SHIFT(p, payloadValuesOffset(p));
+  SMemRow* memRow = (SMemRow)pBuilder->pDataBlock;
+  memRowSetType(memRow, memRowType);
+
+  // ----------------- Raw payload structure for row:
+  /* |<------------ Head ------------->|<----------- body of column data tuple ------------------->|
+   * |                                 |<----------------- flen ------------->|<--- value part --->|
+   * |SMemRowType| dataTLen |  nCols   |  colId  | colType | offset   |  ...  | value |...|...|... |
+   * +-----------+----------+----------+--------------------------------------|--------------------|
+   * | uint8_t   | uint32_t | uint16_t | int16_t | uint8_t | uint16_t |  ...  |.......|...|...|... |
+   * +-----------+----------+----------+--------------------------------------+--------------------|
+   *  1. offset in column data tuple starts from the value part in case of uint16_t overflow.
+   *  2. dataTLen: total length including the header and body.
+   */
+
+  if (memRowType == SMEM_ROW_DATA) {
+    SDataRow trow = (SDataRow)memRowDataBody(memRow);
+    dataRowSetLen(trow, (TDRowLenT)(TD_DATA_ROW_HEAD_SIZE + pBuilder->flen));
+    dataRowSetVersion(trow, pBuilder->sversion);
+
+    p = (char*)payloadBody(pBuilder->buf);
+    uint16_t i = 0, j = 0;
+    while (j < nCols) {
+      if (i >= nColsBound) {
+        break;
+      }
+      int16_t colId = payloadColId(p);
+      if (colId == pSchema[j].colId) {
+        // ASSERT(payloadColType(p) == pSchema[j].type);
+        tdAppendColVal(trow, POINTER_SHIFT(pVals, payloadColOffset(p)), pSchema[j].type, toffset);
+        toffset += TYPE_BYTES[pSchema[j].type];
+        p = payloadNextCol(p);
+        ++i;
+        ++j;
+      } else if (colId < pSchema[j].colId) {
+        p = payloadNextCol(p);
+        ++i;
+      } else {
+        tdAppendColVal(trow, getNullValue(pSchema[j].type), pSchema[j].type, toffset);
+        toffset += TYPE_BYTES[pSchema[j].type];
+        ++j;
+      }
+    }
+
+    while (j < nCols) {
+      tdAppendColVal(trow, getNullValue(pSchema[j].type), pSchema[j].type, toffset);
+      toffset += TYPE_BYTES[pSchema[j].type];
+      ++j;
+    }
+    
+    #if 0 // no need anymore
+    while (i < nColsBound) {
+      p = payloadNextCol(p);
+      ++i;
+    }
+    #endif
+
+  } else if (memRowType == SMEM_ROW_KV) {
+    SKVRow   kvRow = (SKVRow)memRowKvBody(memRow);
+    kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nColsBound));
+    kvRowSetNCols(kvRow, nColsBound);
+    memRowSetKvVersion(memRow, pBuilder->sversion);
+
+    p = (char*)payloadBody(pBuilder->buf);
+    int i = 0;
+    while (i < nColsBound) {
+      int16_t colId = payloadColId(p);
+      uint8_t colType = payloadColType(p);
+      tdAppendKvColVal(kvRow, POINTER_SHIFT(pVals,payloadColOffset(p)), colId, colType, toffset);
+      toffset += sizeof(SColIdx);
+      p = payloadNextCol(p);
+      ++i;
+    }
+
+  } else {
+    ASSERT(0);
+  }
+  int32_t rowTLen = memRowTLen(memRow);
+  pBuilder->pDataBlock = (char*)pBuilder->pDataBlock + rowTLen;  // next row
+  pBuilder->pSubmitBlk->dataLen += rowTLen;
+
+  return memRow;
+}
+
+// Erase the empty space reserved for binary data
+static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, SInsertStatementParam* insertParam,
+                         SBlockKeyTuple* blkKeyTuple) {
   // TODO: optimize this function, handle the case while binary is not presented
-  STableMeta*   pTableMeta = pTableDataBlock->pTableMeta;
-  STableComInfo tinfo = tscGetTableInfo(pTableMeta);
-  SSchema*      pSchema = tscGetTableSchema(pTableMeta);
+  STableMeta*     pTableMeta = pTableDataBlock->pTableMeta;
+  STableComInfo   tinfo = tscGetTableInfo(pTableMeta);
+  SSchema*        pSchema = tscGetTableSchema(pTableMeta);
 
   SSubmitBlk* pBlock = pDataBlock;
   memcpy(pDataBlock, pTableDataBlock->pData, sizeof(SSubmitBlk));
@@ -1775,7 +1873,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
   int32_t flen = 0;  // original total length of row
 
   // schema needs to be included into the submit data block
-  if (includeSchema) {
+  if (insertParam->schemaAttached) {
     int32_t numOfCols = tscGetNumOfColumns(pTableDataBlock->pTableMeta);
     for(int32_t j = 0; j < numOfCols; ++j) {
       STColumn* pCol = (STColumn*) pDataBlock;
@@ -1801,21 +1899,39 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
   char* p = pTableDataBlock->pData + sizeof(SSubmitBlk);
   pBlock->dataLen = 0;
   int32_t numOfRows = htons(pBlock->numOfRows);
-  
-  for (int32_t i = 0; i < numOfRows; ++i) {
-    SDataRow trow = (SDataRow) pDataBlock;
-    dataRowSetLen(trow, (uint16_t)(TD_DATA_ROW_HEAD_SIZE + flen));
-    dataRowSetVersion(trow, pTableMeta->sversion);
-
-    int toffset = 0;
-    for (int32_t j = 0; j < tinfo.numOfColumns; j++) {
-      tdAppendColVal(trow, p, pSchema[j].type, pSchema[j].bytes, toffset);
-      toffset += TYPE_BYTES[pSchema[j].type];
-      p += pSchema[j].bytes;
+
+  if (IS_RAW_PAYLOAD(insertParam->payloadType)) {
+    for (int32_t i = 0; i < numOfRows; ++i) {
+      SMemRow memRow = (SMemRow)pDataBlock;
+      memRowSetType(memRow, SMEM_ROW_DATA);
+      SDataRow trow = memRowDataBody(memRow);
+      dataRowSetLen(trow, (uint16_t)(TD_DATA_ROW_HEAD_SIZE + flen));
+      dataRowSetVersion(trow, pTableMeta->sversion);
+
+      int toffset = 0;
+      for (int32_t j = 0; j < tinfo.numOfColumns; j++) {
+        tdAppendColVal(trow, p, pSchema[j].type, toffset);
+        toffset += TYPE_BYTES[pSchema[j].type];
+        p += pSchema[j].bytes;
+      }
+
+      pDataBlock = (char*)pDataBlock + memRowTLen(memRow);
+      pBlock->dataLen += memRowTLen(memRow);
     }
+  } else {
+    SMemRowBuilder rowBuilder;
+    rowBuilder.pSchema = pSchema;
+    rowBuilder.sversion = pTableMeta->sversion;
+    rowBuilder.flen = flen;
+    rowBuilder.nCols = tinfo.numOfColumns;
+    rowBuilder.pDataBlock = pDataBlock;
+    rowBuilder.pSubmitBlk = pBlock;
+    rowBuilder.buf = p;
 
-    pDataBlock = (char*)pDataBlock + dataRowLen(trow);
-    pBlock->dataLen += dataRowLen(trow);
+    for (int32_t i = 0; i < numOfRows; ++i) {
+      rowBuilder.buf = (blkKeyTuple + i)->payloadAddr;
+      tdGenMemRowFromBuilder(&rowBuilder);
+    }
   }
 
   int32_t len = pBlock->dataLen + pBlock->schemaLen;
@@ -1826,7 +1942,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
 }
 
 static int32_t getRowExpandSize(STableMeta* pTableMeta) {
-  int32_t result = TD_DATA_ROW_HEAD_SIZE;
+  int32_t  result = TD_MEM_ROW_DATA_HEAD_SIZE;
   int32_t columns = tscGetNumOfColumns(pTableMeta);
   SSchema* pSchema = tscGetTableSchema(pTableMeta);
   for(int32_t i = 0; i < columns; i++) {
@@ -1862,13 +1978,17 @@ static void extractTableNameList(SInsertStatementParam *pInsertParam, bool freeB
 
 int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBlockMap) {
   const int INSERT_HEAD_SIZE = sizeof(SMsgDesc) + sizeof(SSubmitMsg);
-
-  void* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
-  SArray* pVnodeDataBlockList = taosArrayInit(8, POINTER_BYTES);
+  int       code = 0;
+  bool      isRawPayload = IS_RAW_PAYLOAD(pInsertParam->payloadType);
+  void*     pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
+  SArray*   pVnodeDataBlockList = taosArrayInit(8, POINTER_BYTES);
 
   STableDataBlocks** p = taosHashIterate(pInsertParam->pTableBlockHashList, NULL);
 
   STableDataBlocks* pOneTableBlock = *p;
+
+  SBlockKeyInfo blkKeyInfo = {0};  // share by pOneTableBlock
+
   while(pOneTableBlock) {
     SSubmitBlk* pBlocks = (SSubmitBlk*) pOneTableBlock->pData;
     if (pBlocks->numOfRows > 0) {
@@ -1882,6 +2002,7 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
         tscError("0x%"PRIx64" failed to prepare the data block buffer for merging table data, code:%d", pInsertParam->objectId, ret);
         taosHashCleanup(pVnodeDataBlockHashList);
         tscDestroyBlockArrayList(pVnodeDataBlockList);
+        tfree(blkKeyInfo.pKeyTuple);
         return ret;
       }
 
@@ -1902,17 +2023,35 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
           taosHashCleanup(pVnodeDataBlockHashList);
           tscDestroyBlockArrayList(pVnodeDataBlockList);
           tfree(dataBuf->pData);
+          tfree(blkKeyInfo.pKeyTuple);
 
           return TSDB_CODE_TSC_OUT_OF_MEMORY;
         }
       }
 
-      tscSortRemoveDataBlockDupRows(pOneTableBlock);
-      char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize*(pBlocks->numOfRows-1);
+      if (isRawPayload) {
+        tscSortRemoveDataBlockDupRowsRaw(pOneTableBlock);
+        char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize * (pBlocks->numOfRows - 1);
 
-      tscDebug("0x%"PRIx64" name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName),
-          pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
+        tscDebug("0x%" PRIx64 " name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64,
+                 pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName), pBlocks->tid,
+                 pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
+      } else {
+        if ((code = tscSortRemoveDataBlockDupRows(pOneTableBlock, &blkKeyInfo)) != 0) {
+          taosHashCleanup(pVnodeDataBlockHashList);
+          tscDestroyBlockArrayList(pVnodeDataBlockList);
+          tfree(dataBuf->pData);
+          tfree(blkKeyInfo.pKeyTuple);
+          return code;
+        }
+        ASSERT(blkKeyInfo.pKeyTuple != NULL && pBlocks->numOfRows > 0);
 
+        SBlockKeyTuple* pLastKeyTuple = blkKeyInfo.pKeyTuple + pBlocks->numOfRows - 1;
+        tscDebug("0x%" PRIx64 " name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64,
+                 pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName), pBlocks->tid,
+                 pBlocks->numOfRows, pBlocks->sversion, blkKeyInfo.pKeyTuple->skey, pLastKeyTuple->skey);
+      }
+      
       int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
 
       pBlocks->tid = htonl(pBlocks->tid);
@@ -1922,7 +2061,7 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
       pBlocks->schemaLen = 0;
 
       // erase the empty space reserved for binary data
-      int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pInsertParam->schemaAttached);
+      int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pInsertParam, blkKeyInfo.pKeyTuple);
       assert(finalLen <= len);
 
       dataBuf->size += (finalLen + sizeof(SSubmitBlk));
@@ -1950,6 +2089,7 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
   // free the table data blocks;
   pInsertParam->pDataBlocks = pVnodeDataBlockList;
   taosHashCleanup(pVnodeDataBlockHashList);
+  tfree(blkKeyInfo.pKeyTuple);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -2986,6 +3126,7 @@ void tscInitQueryInfo(SQueryInfo* pQueryInfo) {
 int32_t tscAddQueryInfo(SSqlCmd* pCmd) {
   assert(pCmd != NULL);
   SQueryInfo* pQueryInfo = calloc(1, sizeof(SQueryInfo));
+
   if (pQueryInfo == NULL) {
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
@@ -3968,17 +4109,21 @@ void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp) {
   //backup the total number of result first
   int64_t num = pRes->numOfTotal + pRes->numOfClauseTotal;
 
-
   // DON't free final since it may be recoreded and used later in APP
   TAOS_FIELD* finalBk = pRes->final;
   pRes->final = NULL;
   tscFreeSqlResult(pSql);
+
   pRes->final = finalBk;
-  
   pRes->numOfTotal = num;
-  
+
+  for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
+    taos_free_result(pSql->pSubs[i]);
+  }
+
   tfree(pSql->pSubs);
   pSql->subState.numOfSub = 0;
+
   pSql->fp = fp;
 
   tscDebug("0x%"PRIx64" try data in the next subclause", pSql->self);
@@ -4618,6 +4763,20 @@ int32_t nameComparFn(const void* n1, const void* n2) {
   }
 }
 
+static void freeContent(void* p) {
+  char* ptr = *(char**)p;
+  tfree(ptr);
+}
+
+static int32_t contCompare(const void* p1, const void* p2) {
+  int32_t ret = strcmp(p1, p2);
+  if (ret == 0) {
+    return 0;
+  } else {
+    return ret > 0 ? 1:-1;
+  }
+}
+
 int tscTransferTableNameList(SSqlObj *pSql, const char *pNameList, int32_t length, SArray* pNameArray) {
   SSqlCmd *pCmd = &pSql->cmd;
 
@@ -4665,32 +4824,7 @@ int tscTransferTableNameList(SSqlObj *pSql, const char *pNameList, int32_t lengt
   }
 
   taosArraySort(pNameArray, nameComparFn);
-
-  int32_t pos = 0;
-  for(int32_t i = 1; i < len; ++i) {
-    char** p1 = taosArrayGet(pNameArray, pos);
-    char** p2 = taosArrayGet(pNameArray, i);
-
-    if (strcmp(*p1, *p2) == 0) {
-      // do nothing
-    } else {
-      if (pos + 1 != i) {
-        char* p = taosArrayGetP(pNameArray, pos + 1);
-        tfree(p);
-        taosArraySet(pNameArray, pos + 1, p2);
-        pos += 1;
-      } else {
-        pos += 1;
-      }
-    }
-  }
-
-  for(int32_t i = pos + 1; i < pNameArray->size; ++i) {
-    char* p = taosArrayGetP(pNameArray, i);
-    tfree(p);
-  }
-
-  pNameArray->size = pos + 1;
+  taosArrayRemoveDuplicate(pNameArray, contCompare, freeContent);
   return TSDB_CODE_SUCCESS;
 }
 

src/common/inc/tdataformat.h

@@ -53,10 +53,10 @@ extern "C" {
 
 // ----------------- TSDB COLUMN DEFINITION
 typedef struct {
-  int8_t  type;    // Column type
-  int16_t colId;   // column ID
-  int16_t bytes;   // column bytes
-  int16_t offset;  // point offset in SDataRow after the header part
+  int8_t   type;    // Column type
+  int16_t  colId;   // column ID
+  uint16_t bytes;   // column bytes
+  uint16_t offset;  // point offset in SDataRow after the header part.
 } STColumn;
 
 #define colType(col) ((col)->type)
@@ -167,10 +167,11 @@ static FORCE_INLINE int tkeyComparFn(const void *tkey1, const void *tkey2) {
     return 0;
   }
 }
+
 // ----------------- Data row structure
 
 /* A data row, the format is like below:
- * |<--------------------+--------------------------- len ---------------------------------->|
+ * |<------------------------------------------------ len ---------------------------------->|
  * |<--     Head      -->|<---------   flen -------------->|                                 |
  * +---------------------+---------------------------------+---------------------------------+
  * | uint16_t |  int16_t |                                 |                                 |
@@ -184,8 +185,8 @@ typedef void *SDataRow;
 
 #define TD_DATA_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
 
-#define dataRowLen(r) (*(uint16_t *)(r))
-#define dataRowVersion(r) *(int16_t *)POINTER_SHIFT(r, sizeof(int16_t))
+#define dataRowLen(r) (*(TDRowLenT *)(r))  // 0~65535
+#define dataRowVersion(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
 #define dataRowTuple(r) POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
 #define dataRowTKey(r) (*(TKEY *)(dataRowTuple(r)))
 #define dataRowKey(r) tdGetKey(dataRowTKey(r))
@@ -201,20 +202,19 @@ void     tdInitDataRow(SDataRow row, STSchema *pSchema);
 SDataRow tdDataRowDup(SDataRow row);
 
 // offset here not include dataRow header length
-static FORCE_INLINE int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_t offset) {
+static FORCE_INLINE int tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
   ASSERT(value != NULL);
   int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
-  char *  ptr = (char *)POINTER_SHIFT(row, dataRowLen(row));
 
   if (IS_VAR_DATA_TYPE(type)) {
     *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
-    memcpy(ptr, value, varDataTLen(value));
+    memcpy(POINTER_SHIFT(row, dataRowLen(row)), value, varDataTLen(value));
     dataRowLen(row) += varDataTLen(value);
   } else {
     if (offset == 0) {
       ASSERT(type == TSDB_DATA_TYPE_TIMESTAMP);
       TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
-      memcpy(POINTER_SHIFT(row, toffset), (void *)(&tvalue), TYPE_BYTES[type]);
+      memcpy(POINTER_SHIFT(row, toffset), (const void *)(&tvalue), TYPE_BYTES[type]);
     } else {
       memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
     }
@@ -245,17 +245,21 @@ typedef struct SDataCol {
   TSKEY           ts;         // only used in last NULL column
 } SDataCol;
 
+#define isAllRowsNull(pCol) ((pCol)->len == 0)
 static FORCE_INLINE void dataColReset(SDataCol *pDataCol) { pDataCol->len = 0; }
 
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints);
-void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints);
+void dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints);
 void dataColSetOffset(SDataCol *pCol, int nEle);
 
 bool isNEleNull(SDataCol *pCol, int nEle);
 void dataColSetNEleNull(SDataCol *pCol, int nEle, int maxPoints);
 
 // Get the data pointer from a column-wised data
-static FORCE_INLINE void *tdGetColDataOfRow(SDataCol *pCol, int row) {
+static FORCE_INLINE const void *tdGetColDataOfRow(SDataCol *pCol, int row) {
+  if (isAllRowsNull(pCol)) {
+    return getNullValue(pCol->type);
+  }
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
   } else {
@@ -287,7 +291,7 @@ typedef struct {
 } SDataCols;
 
 #define keyCol(pCols) (&((pCols)->cols[0]))  // Key column
-#define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]
+#define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]  // the idx row of column-wised data
 #define dataColsKeyAt(pCols, idx) tdGetKey(dataColsTKeyAt(pCols, idx))
 static FORCE_INLINE TKEY dataColsTKeyFirst(SDataCols *pCols) {
   if (pCols->numOfRows) {
@@ -331,13 +335,13 @@ void       tdResetDataCols(SDataCols *pCols);
 int        tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 SDataCols *tdFreeDataCols(SDataCols *pCols);
-void       tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols);
 int        tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset);
 
 // ----------------- K-V data row structure
-/*
+/* |<-------------------------------------- len -------------------------------------------->|
+ * |<----- header  ----->|<--------------------------- body -------------------------------->|
  * +----------+----------+---------------------------------+---------------------------------+
- * |  int16_t |  int16_t |                                 |                                 |
+ * | uint16_t |  int16_t |                                 |                                 |
  * +----------+----------+---------------------------------+---------------------------------+
  * |    len   |   ncols  |           cols index            |             data part           |
  * +----------+----------+---------------------------------+---------------------------------+
@@ -345,14 +349,14 @@ int        tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge
 typedef void *SKVRow;
 
 typedef struct {
-  int16_t colId;
-  int16_t offset;
+  int16_t  colId;
+  uint16_t offset;
 } SColIdx;
 
-#define TD_KV_ROW_HEAD_SIZE (2 * sizeof(int16_t))
+#define TD_KV_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
 
-#define kvRowLen(r) (*(int16_t *)(r))
-#define kvRowNCols(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
+#define kvRowLen(r) (*(TDRowLenT *)(r))
+#define kvRowNCols(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(uint16_t)))
 #define kvRowSetLen(r, len) kvRowLen(r) = (len)
 #define kvRowSetNCols(r, n) kvRowNCols(r) = (n)
 #define kvRowColIdx(r) (SColIdx *)POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE)
@@ -362,6 +366,9 @@ typedef struct {
 #define kvRowColIdxAt(r, i) (kvRowColIdx(r) + (i))
 #define kvRowFree(r) tfree(r)
 #define kvRowEnd(r) POINTER_SHIFT(r, kvRowLen(r))
+#define kvRowTKey(r) (*(TKEY *)(kvRowValues(r)))
+#define kvRowKey(r) tdGetKey(kvRowTKey(r))
+#define kvRowDeleted(r) TKEY_IS_DELETED(kvRowTKey(r))
 
 SKVRow tdKVRowDup(SKVRow row);
 int    tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value);
@@ -385,13 +392,44 @@ static FORCE_INLINE void *tdGetKVRowValOfCol(SKVRow row, int16_t colId) {
   return kvRowColVal(row, (SColIdx *)ret);
 }
 
+static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
+  return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
+}
+
+// offset here not include kvRow header length
+static FORCE_INLINE int tdAppendKvColVal(SKVRow row, const void *value, int16_t colId, int8_t type, int32_t offset) {
+  ASSERT(value != NULL);
+  int32_t  toffset = offset + TD_KV_ROW_HEAD_SIZE;
+  SColIdx *pColIdx = (SColIdx *)POINTER_SHIFT(row, toffset);
+  char *   ptr = (char *)POINTER_SHIFT(row, kvRowLen(row));
+
+  pColIdx->colId = colId;
+  pColIdx->offset = kvRowLen(row);  // offset of pColIdx including the TD_KV_ROW_HEAD_SIZE
+
+  if (IS_VAR_DATA_TYPE(type)) {
+    memcpy(ptr, value, varDataTLen(value));
+    kvRowLen(row) += varDataTLen(value);
+  } else {
+    if (offset == 0) {
+      ASSERT(type == TSDB_DATA_TYPE_TIMESTAMP);
+      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
+      memcpy(ptr, (void *)(&tvalue), TYPE_BYTES[type]);
+    } else {
+      memcpy(ptr, value, TYPE_BYTES[type]);
+    }
+    kvRowLen(row) += TYPE_BYTES[type];
+  }
+
+  return 0;
+}
+
 // ----------------- K-V data row builder
 typedef struct {
   int16_t  tCols;
   int16_t  nCols;
   SColIdx *pColIdx;
-  int16_t  alloc;
-  int16_t  size;
+  uint16_t alloc;
+  uint16_t size;
   void *   buf;
 } SKVRowBuilder;
 
@@ -427,8 +465,146 @@ static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId,
   return 0;
 }
 
+// ----------------- SMemRow appended with sequential data row structure
+/*
+ * |---------|------------------------------------------------- len ---------------------------------->|
+ * |<--------     Head      ------>|<---------   flen -------------->|                                 |
+ * |---------+---------------------+---------------------------------+---------------------------------+
+ * | uint8_t | uint16_t |  int16_t |                                 |                                 |
+ * |---------+----------+----------+---------------------------------+---------------------------------+
+ * |  flag   |   len    | sversion |           First part            |             Second part         |
+ * +---------+----------+----------+---------------------------------+---------------------------------+
+ *
+ * NOTE: timestamp in this row structure is TKEY instead of TSKEY
+ */
+
+// ----------------- SMemRow appended with extended K-V data row structure
+/* |--------------------|------------------------------------------------  len ---------------------------------->|
+ * |<-------------     Head      ------------>|<---------   flen -------------->|                                 |
+ * |--------------------+----------+--------------------------------------------+---------------------------------+
+ * | uint8_t | int16_t  | uint16_t |  int16_t |                                 |                                 |
+ * |---------+----------+----------+----------+---------------------------------+---------------------------------+
+ * |   flag  | sversion |   len    |   ncols  |           cols index            |             data part           |
+ * |---------+----------+----------+----------+---------------------------------+---------------------------------+
+ */
+
+typedef void *SMemRow;
+
+#define TD_MEM_ROW_TYPE_SIZE sizeof(uint8_t)
+#define TD_MEM_ROW_KV_VER_SIZE sizeof(int16_t)
+#define TD_MEM_ROW_KV_TYPE_VER_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE)
+#define TD_MEM_ROW_DATA_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_DATA_ROW_HEAD_SIZE)
+// #define TD_MEM_ROW_KV_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE + TD_KV_ROW_HEAD_SIZE)
+
+#define SMEM_ROW_DATA 0U  // SDataRow
+#define SMEM_ROW_KV 1U    // SKVRow
+
+#define memRowType(r) (*(uint8_t *)(r))
+#define isDataRow(r) (SMEM_ROW_DATA == memRowType(r))
+#define isKvRow(r) (SMEM_ROW_KV == memRowType(r))
+
+#define memRowDataBody(r) POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE)  // section after flag
+#define memRowKvBody(r) \
+  POINTER_SHIFT(r, TD_MEM_ROW_KV_TYPE_VER_SIZE)  // section after flag + sversion as to reuse SKVRow
+
+#define memRowDataLen(r) (*(TDRowLenT *)memRowDataBody(r))  //  0~65535
+#define memRowKvLen(r) (*(TDRowLenT *)memRowKvBody(r))      //  0~65535
+
+#define memRowDataTLen(r) \
+  ((TDRowTLenT)(memRowDataLen(r) + TD_MEM_ROW_TYPE_SIZE))  // using uint32_t/int32_t to store the TLen
+
+#define memRowKvTLen(r) ((TDRowTLenT)(memRowKvLen(r) + TD_MEM_ROW_KV_TYPE_VER_SIZE))
+
+#define memRowLen(r) (isDataRow(r) ? memRowDataLen(r) : memRowKvLen(r))
+#define memRowTLen(r) (isDataRow(r) ? memRowDataTLen(r) : memRowKvTLen(r))  // using uint32_t/int32_t to store the TLen
+
+#define memRowDataVersion(r) dataRowVersion(memRowDataBody(r))
+#define memRowKvVersion(r) (*(int16_t *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))
+#define memRowVersion(r) (isDataRow(r) ? memRowDataVersion(r) : memRowKvVersion(r))  // schema version
+#define memRowSetKvVersion(r, v) (memRowKvVersion(r) = (v))
+#define memRowTuple(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowValues(memRowKvBody(r)))
+
+#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
+#define memRowKey(r) (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
+#define memRowSetTKey(r, k)                 \
+  do {                                      \
+    if (isDataRow(r)) {                     \
+      dataRowTKey(memRowDataBody(r)) = (k); \
+    } else {                                \
+      kvRowTKey(memRowKvBody(r)) = (k);     \
+    }                                       \
+  } while (0)
+
+#define memRowSetType(r, t) (memRowType(r) = (t))
+#define memRowSetLen(r, l) (isDataRow(r) ? memRowDataLen(r) = (l) : memRowKvLen(r) = (l))
+#define memRowSetVersion(r, v) (isDataRow(r) ? dataRowSetVersion(memRowDataBody(r), v) : memRowKvSetVersion(r, v))
+#define memRowCpy(dst, r) memcpy((dst), (r), memRowTLen(r))
+#define memRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_MEM_ROW_DATA_HEAD_SIZE)
+#define memRowDeleted(r) TKEY_IS_DELETED(memRowTKey(r))
+
+SMemRow tdMemRowDup(SMemRow row);
+void    tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols);
+// NOTE: offset here including the header size
+static FORCE_INLINE void *tdGetKvRowDataOfCol(void *row, int32_t offset) { return POINTER_SHIFT(row, offset); }
+// NOTE: offset here including the header size
+static FORCE_INLINE void *tdGetMemRowDataOfCol(void *row, int8_t type, int32_t offset) {
+  if (isDataRow(row)) {
+    return tdGetRowDataOfCol(row, type, offset);
+  } else if (isKvRow(row)) {
+    return tdGetKvRowDataOfCol(row, offset);
+  } else {
+    ASSERT(0);
+  }
+  return NULL;
+}
+
+// ----------------- Raw payload structure for row:
+/* |<------------ Head ------------->|<----------- body of column data tuple ------------------->|
+ * |                                 |<----------------- flen ------------->|<--- value part --->|
+ * |SMemRowType| dataTLen |  nCols   |  colId  | colType | offset   |  ...  | value |...|...|... |
+ * +-----------+----------+----------+--------------------------------------|--------------------|
+ * | uint8_t   | uint32_t | uint16_t | int16_t | uint8_t | uint16_t |  ...  |.......|...|...|... |
+ * +-----------+----------+----------+--------------------------------------+--------------------|
+ *  1. offset in column data tuple starts from the value part in case of uint16_t overflow.
+ *  2. dataTLen: total length including the header and body.
+ */
+
+#define PAYLOAD_NCOLS_LEN sizeof(uint16_t)
+#define PAYLOAD_NCOLS_OFFSET (sizeof(uint8_t) + sizeof(TDRowTLenT))
+#define PAYLOAD_HEADER_LEN (PAYLOAD_NCOLS_OFFSET + PAYLOAD_NCOLS_LEN)
+#define PAYLOAD_ID_LEN sizeof(int16_t)
+#define PAYLOAD_ID_TYPE_LEN (sizeof(int16_t) + sizeof(uint8_t))
+#define PAYLOAD_COL_HEAD_LEN (PAYLOAD_ID_TYPE_LEN + sizeof(uint16_t))
+#define PAYLOAD_PRIMARY_COL_LEN (PAYLOAD_ID_TYPE_LEN + sizeof(TSKEY))
+
+#define payloadBody(r) POINTER_SHIFT(r, PAYLOAD_HEADER_LEN)
+#define payloadType(r) (*(uint8_t *)(r))
+#define payloadSetType(r, t) (payloadType(r) = (t))
+#define payloadTLen(r) (*(TDRowTLenT *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))  // including total header
+#define payloadSetTLen(r, l) (payloadTLen(r) = (l))
+#define payloadNCols(r) (*(TDRowLenT *)POINTER_SHIFT(r, PAYLOAD_NCOLS_OFFSET))
+#define payloadSetNCols(r, n) (payloadNCols(r) = (n))
+#define payloadValuesOffset(r) \
+  (PAYLOAD_HEADER_LEN + payloadNCols(r) * PAYLOAD_COL_HEAD_LEN)    // avoid using the macro in loop
+#define payloadValues(r) POINTER_SHIFT(r, payloadValuesOffset(r))  // avoid using the macro in loop
+#define payloadColId(c) (*(int16_t *)(c))
+#define payloadColType(c) (*(uint8_t *)POINTER_SHIFT(c, PAYLOAD_ID_LEN))
+#define payloadColOffset(c) (*(uint16_t *)POINTER_SHIFT(c, PAYLOAD_ID_TYPE_LEN))
+#define payloadColValue(c) POINTER_SHIFT(c, payloadColOffset(c))
+
+#define payloadColSetId(c, i) (payloadColId(c) = (i))
+#define payloadColSetType(c, t) (payloadColType(c) = (t))
+#define payloadColSetOffset(c, o) (payloadColOffset(c) = (o))
+
+#define payloadTSKey(r) (*(TSKEY *)POINTER_SHIFT(r, payloadValuesOffset(r)))
+#define payloadTKey(r) (*(TKEY *)POINTER_SHIFT(r, payloadValuesOffset(r)))
+#define payloadKey(r) tdGetKey(payloadTKey(r))
+
+
+static FORCE_INLINE char *payloadNextCol(char *pCol) { return (char *)POINTER_SHIFT(pCol, PAYLOAD_COL_HEAD_LEN); }
+
 #ifdef __cplusplus
 }
 #endif
 
-#endif  // _TD_DATA_FORMAT_H_
+#endif  // _TD_DATA_FORMAT_H_
\ No newline at end of file

src/common/src/tdataformat.c

@@ -198,6 +198,14 @@ SDataRow tdDataRowDup(SDataRow row) {
   return trow;
 }
 
+SMemRow tdMemRowDup(SMemRow row) {
+  SMemRow trow = malloc(memRowTLen(row));
+  if (trow == NULL) return NULL;
+
+  memRowCpy(trow, row);
+  return trow;
+}
+
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints) {
   pDataCol->type = colType(pCol);
   pDataCol->colId = colColId(pCol);
@@ -217,11 +225,22 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
     *pBuf = POINTER_SHIFT(*pBuf, pDataCol->spaceSize);
   }
 }
-
 // value from timestamp should be TKEY here instead of TSKEY
-void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints) {
+void dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints) {
   ASSERT(pCol != NULL && value != NULL);
 
+  if (isAllRowsNull(pCol)) {
+    if (isNull(value, pCol->type)) {
+      // all null value yet, just return
+      return;
+    }
+
+    if (numOfRows > 0) {
+      // Find the first not null value, fill all previouse values as NULL
+      dataColSetNEleNull(pCol, numOfRows, maxPoints);
+    }
+  }
+
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     // set offset
     pCol->dataOff[numOfRows] = pCol->len;
@@ -243,7 +262,7 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
   return true;
 }
 
-void dataColSetNullAt(SDataCol *pCol, int index) {
+FORCE_INLINE void dataColSetNullAt(SDataCol *pCol, int index) {
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     pCol->dataOff[index] = pCol->len;
     char *ptr = POINTER_SHIFT(pCol->pData, pCol->len);
@@ -399,8 +418,7 @@ void tdResetDataCols(SDataCols *pCols) {
     }
   }
 }
-
-void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols) {
+static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols) {
   ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
 
   int rcol = 0;
@@ -419,7 +437,8 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
     while (dcol < pCols->numOfCols) {
       SDataCol *pDataCol = &(pCols->cols[dcol]);
       if (rcol >= schemaNCols(pSchema)) {
-        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
         dcol++;
         continue;
       }
@@ -433,7 +452,8 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
       } else if (pRowCol->colId < pDataCol->colId) {
         rcol++;
       } else {
-        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
         dcol++;
       }
     }
@@ -441,6 +461,62 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
   pCols->numOfRows++;
 }
 
+static void tdAppendKvRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCols) {
+  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < kvRowKey(row));
+
+  int rcol = 0;
+  int dcol = 0;
+
+  if (kvRowDeleted(row)) {
+    for (; dcol < pCols->numOfCols; dcol++) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (dcol == 0) {
+        dataColAppendVal(pDataCol, kvRowValues(row), pCols->numOfRows, pCols->maxPoints);
+      } else {
+        dataColSetNullAt(pDataCol, pCols->numOfRows);
+      }
+    }
+  } else {
+    int nRowCols = kvRowNCols(row);
+
+    while (dcol < pCols->numOfCols) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+        continue;
+      }
+
+      SColIdx *colIdx = kvRowColIdxAt(row, rcol);
+
+      if (colIdx->colId == pDataCol->colId) {
+        void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
+        dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+        ++rcol;
+      } else if (colIdx->colId < pDataCol->colId) {
+        ++rcol;
+      } else {
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+      }
+    }
+  }
+  pCols->numOfRows++;
+}
+
+void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols) {
+  if (isDataRow(row)) {
+    tdAppendDataRowToDataCol(memRowDataBody(row), pSchema, pCols);
+  } else if (isKvRow(row)) {
+    tdAppendKvRowToDataCol(memRowKvBody(row), pSchema, pCols);
+  } else {
+    ASSERT(0);
+  }
+}
+
 int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset) {
   ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
   ASSERT(target->numOfCols == source->numOfCols);
@@ -563,7 +639,7 @@ int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
     nrow = malloc(kvRowLen(row) + sizeof(SColIdx) + diff);
     if (nrow == NULL) return -1;
 
-    kvRowSetLen(nrow, kvRowLen(row) + (int16_t)sizeof(SColIdx) + diff);
+    kvRowSetLen(nrow, kvRowLen(row) + (uint16_t)sizeof(SColIdx) + diff);
     kvRowSetNCols(nrow, kvRowNCols(row) + 1);
 
     if (ptr == NULL) {
@@ -605,8 +681,8 @@ int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
       if (varDataTLen(value) == varDataTLen(pOldVal)) { // just update the column value in place
         memcpy(pOldVal, value, varDataTLen(value));
       } else { // need to reallocate the memory
-        int16_t diff = varDataTLen(value) - varDataTLen(pOldVal);
-        int16_t nlen = kvRowLen(row) + diff;
+        uint16_t diff = varDataTLen(value) - varDataTLen(pOldVal);
+        uint16_t nlen = kvRowLen(row) + diff;
         ASSERT(nlen > 0);
         nrow = malloc(nlen);
         if (nrow == NULL) return -1;
@@ -708,4 +784,4 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
   memcpy(kvRowValues(row), pBuilder->buf, pBuilder->size);
 
   return row;
-}
+}
\ No newline at end of file

src/common/src/ttypes.c

@@ -492,30 +492,32 @@ void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems) {
   }
 }
 
-static uint8_t   nullBool     = TSDB_DATA_BOOL_NULL;
-static uint8_t   nullTinyInt  = TSDB_DATA_TINYINT_NULL;
-static uint16_t  nullSmallInt = TSDB_DATA_SMALLINT_NULL;
-static uint32_t  nullInt      = TSDB_DATA_INT_NULL;
-static uint64_t  nullBigInt   = TSDB_DATA_BIGINT_NULL;
-static uint32_t  nullFloat    = TSDB_DATA_FLOAT_NULL;
-static uint64_t  nullDouble   = TSDB_DATA_DOUBLE_NULL;
-static uint8_t   nullTinyIntu = TSDB_DATA_UTINYINT_NULL;
-static uint16_t  nullSmallIntu = TSDB_DATA_USMALLINT_NULL;
-static uint32_t  nullIntu     = TSDB_DATA_UINT_NULL;
-static uint64_t  nullBigIntu  = TSDB_DATA_UBIGINT_NULL;
-
-static union {
-  tstr str;
-  char pad[sizeof(tstr) + 4];
-} nullBinary = {.str = {.len = 1}}, nullNchar = {.str = {.len = 4}};
-
-static void *nullValues[] = {
+static uint8_t      nullBool = TSDB_DATA_BOOL_NULL;
+static uint8_t      nullTinyInt = TSDB_DATA_TINYINT_NULL;
+static uint16_t     nullSmallInt = TSDB_DATA_SMALLINT_NULL;
+static uint32_t     nullInt = TSDB_DATA_INT_NULL;
+static uint64_t     nullBigInt = TSDB_DATA_BIGINT_NULL;
+static uint32_t     nullFloat = TSDB_DATA_FLOAT_NULL;
+static uint64_t     nullDouble = TSDB_DATA_DOUBLE_NULL;
+static uint8_t      nullTinyIntu = TSDB_DATA_UTINYINT_NULL;
+static uint16_t     nullSmallIntu = TSDB_DATA_USMALLINT_NULL;
+static uint32_t     nullIntu = TSDB_DATA_UINT_NULL;
+static uint64_t     nullBigIntu = TSDB_DATA_UBIGINT_NULL;
+static SBinaryNullT nullBinary = {1, TSDB_DATA_BINARY_NULL};
+static SNCharNullT  nullNchar = {4, TSDB_DATA_NCHAR_NULL};
+
+// static union {
+//   tstr str;
+//   char pad[sizeof(tstr) + 4];
+// } nullBinary = {.str = {.len = 1}}, nullNchar = {.str = {.len = 4}};
+
+static const void *nullValues[] = {
     &nullBool,     &nullTinyInt,   &nullSmallInt, &nullInt,    &nullBigInt,
     &nullFloat,    &nullDouble,    &nullBinary,   &nullBigInt, &nullNchar,
     &nullTinyIntu, &nullSmallIntu, &nullIntu,     &nullBigIntu,
 };
 
-void *getNullValue(int32_t type) {
+const void *getNullValue(int32_t type) {
   assert(type >= TSDB_DATA_TYPE_BOOL && type <= TSDB_DATA_TYPE_UBIGINT);
   return nullValues[type - 1];
 }

src/cq/src/cqMain.c

@@ -476,21 +476,23 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   
   cDebug("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
 
-  int32_t size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize;
+  int32_t size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_MEM_ROW_DATA_HEAD_SIZE + pObj->rowSize;
   char *buffer = calloc(size, 1);
 
   SWalHead   *pHead = (SWalHead *)buffer;
   SSubmitMsg *pMsg = (SSubmitMsg *) (buffer + sizeof(SWalHead));
   SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
 
-  SDataRow trow = (SDataRow)pBlk->data;
-  tdInitDataRow(trow, pSchema);
+  SMemRow trow = (SMemRow)pBlk->data;
+  SDataRow dataRow = (SDataRow)memRowDataBody(trow);
+  memRowSetType(trow, SMEM_ROW_DATA);
+  tdInitDataRow(dataRow, pSchema);
 
   for (int32_t i = 0; i < pSchema->numOfCols; i++) {
     STColumn *c = pSchema->columns + i;
-    void* val = row[i];
+    void *val = row[i];
     if (val == NULL) {
-      val = getNullValue(c->type);
+      val = (void *)getNullValue(c->type);
     } else if (c->type == TSDB_DATA_TYPE_BINARY) {
       val = ((char*)val) - sizeof(VarDataLenT);
     } else if (c->type == TSDB_DATA_TYPE_NCHAR) {
@@ -500,9 +502,9 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
       memcpy((char *)val + sizeof(VarDataLenT), buf, len);
       varDataLen(val) = len;
     }
-    tdAppendColVal(trow, val, c->type, c->bytes, c->offset);
+    tdAppendColVal(dataRow, val, c->type, c->offset);
   }
-  pBlk->dataLen = htonl(dataRowLen(trow));
+  pBlk->dataLen = htonl(memRowDataTLen(trow));
   pBlk->schemaLen = 0;
 
   pBlk->uid = htobe64(pObj->uid);
@@ -511,7 +513,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   pBlk->sversion = htonl(pSchema->version);
   pBlk->padding = 0;
 
-  pHead->len = sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + dataRowLen(trow);
+  pHead->len = sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + memRowDataTLen(trow);
 
   pMsg->header.vgId = htonl(pContext->vgId);
   pMsg->header.contLen = htonl(pHead->len);

src/inc/taos.h

@@ -72,6 +72,7 @@ DLL_EXPORT int  taos_init();
 DLL_EXPORT void  taos_cleanup(void);
 DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
 DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
+DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
 DLL_EXPORT void  taos_close(TAOS *taos);
 
 const char *taos_data_type(int type);
@@ -110,21 +111,21 @@ typedef struct TAOS_MULTI_BIND {
 } TAOS_MULTI_BIND;
 
 
-TAOS_STMT *taos_stmt_init(TAOS *taos);
-int        taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
-int        taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags);
-int        taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name);
-int        taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
-int        taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
+DLL_EXPORT TAOS_STMT *taos_stmt_init(TAOS *taos);
+DLL_EXPORT int        taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
+DLL_EXPORT int        taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags);
+DLL_EXPORT int        taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name);
+DLL_EXPORT int        taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
+DLL_EXPORT int        taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
 int        taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes);
-int        taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_BIND *bind);
+DLL_EXPORT int        taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_BIND *bind);
 int        taos_stmt_bind_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind);
 int        taos_stmt_bind_single_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int colIdx);
-int        taos_stmt_add_batch(TAOS_STMT *stmt);
-int        taos_stmt_execute(TAOS_STMT *stmt);
-TAOS_RES * taos_stmt_use_result(TAOS_STMT *stmt);
-int        taos_stmt_close(TAOS_STMT *stmt);
-char *     taos_stmt_errstr(TAOS_STMT *stmt);
+DLL_EXPORT int        taos_stmt_add_batch(TAOS_STMT *stmt);
+DLL_EXPORT int        taos_stmt_execute(TAOS_STMT *stmt);
+DLL_EXPORT TAOS_RES * taos_stmt_use_result(TAOS_STMT *stmt);
+DLL_EXPORT int        taos_stmt_close(TAOS_STMT *stmt);
+DLL_EXPORT char *     taos_stmt_errstr(TAOS_STMT *stmt);
 
 DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
 DLL_EXPORT TAOS_ROW taos_fetch_row(TAOS_RES *res);
@@ -139,10 +140,10 @@ DLL_EXPORT int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int n
 DLL_EXPORT void taos_stop_query(TAOS_RES *res);
 DLL_EXPORT bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col);
 
-int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows);
-int taos_validate_sql(TAOS *taos, const char *sql);
+DLL_EXPORT int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows);
+DLL_EXPORT int taos_validate_sql(TAOS *taos, const char *sql);
 
-int* taos_fetch_lengths(TAOS_RES *res);
+DLL_EXPORT int* taos_fetch_lengths(TAOS_RES *res);
 
 // TAOS_RES   *taos_list_tables(TAOS *mysql, const char *wild);
 // TAOS_RES   *taos_list_dbs(TAOS *mysql, const char *wild);

src/inc/taosdef.h

@@ -180,7 +180,7 @@ do { \
 // this is the length of its string representation, including the terminator zero
 #define TSDB_ACCT_ID_LEN          11
 
-#define TSDB_MAX_COLUMNS          1024
+#define TSDB_MAX_COLUMNS          4096
 #define TSDB_MIN_COLUMNS          2       //PRIMARY COLUMN(timestamp) + other columns
 
 #define TSDB_NODE_NAME_LEN        64
@@ -199,7 +199,13 @@ do { \
 
 #define TSDB_APPNAME_LEN          TSDB_UNI_LEN
 
-#define TSDB_MAX_BYTES_PER_ROW    16384
+  /**
+   *  In some scenarios uint16_t (0~65535) is used to store the row len.
+   *  - Firstly, we use 65531(65535 - 4), as the SDataRow/SKVRow contains 4 bits header.
+   *  - Secondly, if all cols are VarDataT type except primary key, we need 4 bits to store the offset, thus
+   *    the final value is 65531-(4096-1)*4 = 49151.
+   */
+#define TSDB_MAX_BYTES_PER_ROW    49151
 #define TSDB_MAX_TAGS_LEN         16384
 #define TSDB_MAX_TAGS             128
 #define TSDB_MAX_TAG_CONDITIONS   1024
@@ -327,8 +333,9 @@ do { \
 #define TSDB_MAX_JOIN_TABLE_NUM         10
 #define TSDB_MAX_UNION_CLAUSE           5
 
-#define TSDB_MAX_BINARY_LEN            (TSDB_MAX_BYTES_PER_ROW-TSDB_KEYSIZE)
-#define TSDB_MAX_NCHAR_LEN             (TSDB_MAX_BYTES_PER_ROW-TSDB_KEYSIZE)
+#define TSDB_MAX_FIELD_LEN              16384
+#define TSDB_MAX_BINARY_LEN            (TSDB_MAX_FIELD_LEN-TSDB_KEYSIZE) // keep 16384
+#define TSDB_MAX_NCHAR_LEN             (TSDB_MAX_FIELD_LEN-TSDB_KEYSIZE) // keep 16384
 #define PRIMARYKEY_TIMESTAMP_COL_INDEX  0
 
 #define TSDB_MAX_RPC_THREADS            5

src/inc/taoserror.h

@@ -174,6 +174,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_FIELD_NOT_EXIST           TAOS_DEF_ERROR_CODE(0, 0x036C)  //"Field does not exist")
 #define TSDB_CODE_MND_INVALID_STABLE_NAME       TAOS_DEF_ERROR_CODE(0, 0x036D)  //"Super table does not exist")
 #define TSDB_CODE_MND_INVALID_CREATE_TABLE_MSG  TAOS_DEF_ERROR_CODE(0, 0x036E)  //"Invalid create table message")
+#define TSDB_CODE_MND_EXCEED_MAX_ROW_BYTES      TAOS_DEF_ERROR_CODE(0, 0x036F)  //"Exceed max row bytes")
 
 #define TSDB_CODE_MND_INVALID_FUNC_NAME         TAOS_DEF_ERROR_CODE(0, 0x0370)  //"Invalid func name")
 #define TSDB_CODE_MND_INVALID_FUNC_LEN          TAOS_DEF_ERROR_CODE(0, 0x0371)  //"Invalid func length")

src/inc/ttype.h

@@ -10,14 +10,28 @@ extern "C" {
 #include "taosdef.h"
 
 // ----------------- For variable data types such as TSDB_DATA_TYPE_BINARY and TSDB_DATA_TYPE_NCHAR
-typedef int32_t VarDataOffsetT;
-typedef int16_t VarDataLenT;
+typedef int32_t  VarDataOffsetT;
+typedef int16_t  VarDataLenT;  // maxVarDataLen: 32767
+typedef uint16_t TDRowLenT;    // not including overhead: 0 ~ 65535
+typedef uint32_t TDRowTLenT;   // total length, including overhead
 
 typedef struct tstr {
   VarDataLenT len;
   char        data[];
 } tstr;
 
+#pragma pack(push, 1)
+typedef struct {
+  VarDataLenT len;
+  uint8_t     data;
+} SBinaryNullT;
+
+typedef struct {
+  VarDataLenT len;
+  uint32_t    data;
+} SNCharNullT;
+#pragma pack(pop)
+
 #define VARSTR_HEADER_SIZE  sizeof(VarDataLenT)
 
 #define varDataLen(v)       ((VarDataLenT *)(v))[0]
@@ -180,7 +194,7 @@ bool isValidDataType(int32_t type);
 void  setVardataNull(char* val, int32_t type);
 void  setNull(char *val, int32_t type, int32_t bytes);
 void  setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems);
-void *getNullValue(int32_t type);
+const void *getNullValue(int32_t type);
 
 void assignVal(char *val, const char *src, int32_t len, int32_t type);
 void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf);

src/mnode/src/mnodeTable.c

@@ -1503,6 +1503,18 @@ static int32_t mnodeChangeSuperTableColumn(SMnodeMsg *pMsg) {
     return TSDB_CODE_MND_FIELD_NOT_EXIST;
   }
 
+  // check exceed max row bytes
+  int32_t i;
+  uint32_t nLen = 0;
+  for (i = 0; i < pStable->numOfColumns; ++i) {
+    nLen += (pStable->schema[i].colId == col) ? pAlter->schema[0].bytes : pStable->schema[i].bytes;
+  }
+  if (nLen > TSDB_MAX_BYTES_PER_ROW) {
+    mError("msg:%p, app:%p stable:%s, change column, name:%s exceed max row bytes", pMsg, pMsg->rpcMsg.ahandle,
+           pStable->info.tableId, name);
+    return TSDB_CODE_MND_EXCEED_MAX_ROW_BYTES;
+  }
+
   // update
   SSchema *schema = (SSchema *) (pStable->schema + col);
   ASSERT(schema->type == TSDB_DATA_TYPE_BINARY || schema->type == TSDB_DATA_TYPE_NCHAR);

src/plugins/http/inc/httpParser.h

@@ -100,6 +100,7 @@ typedef enum HTTP_PARSER_STATE {
   HTTP_PARSER_CHUNK,
   HTTP_PARSER_END,
   HTTP_PARSER_ERROR,
+  HTTP_PARSER_OPTIONAL_SP
 } HTTP_PARSER_STATE;
 
 typedef enum HTTP_AUTH_TYPE {

src/plugins/http/src/httpParser.c

@@ -744,6 +744,15 @@ static int32_t httpParserOnSp(HttpParser *parser, HTTP_PARSER_STATE state, const
   return ok;
 }
 
+static int32_t httpParserOnOptionalSp(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) {
+  int32_t ok = 0;
+  if (c != ' ') {
+    *again = 1;
+    httpPopStack(parser);
+  }
+  return ok;
+}
+
 static int32_t httpParserOnStatusCode(HttpParser *parser, HTTP_PARSER_STATE state, const char c, int32_t *again) {
   HttpContext *pContext = parser->pContext;
   int32_t      ok = 0;
@@ -867,7 +876,7 @@ static int32_t httpParserOnHeader(HttpParser *parser, HTTP_PARSER_STATE state, c
       }
       httpPushStack(parser, HTTP_PARSER_CRLF);
       httpPushStack(parser, HTTP_PARSER_HEADER_VAL);
-      httpPushStack(parser, HTTP_PARSER_SP);
+      httpPushStack(parser, HTTP_PARSER_OPTIONAL_SP);
       httpPushStack(parser, HTTP_PARSER_HEADER_KEY);
       break;
     }
@@ -1061,6 +1070,10 @@ static int32_t httpParseChar(HttpParser *parser, const char c, int32_t *again) {
       ok = httpParserOnSp(parser, state, c, again);
       break;
     }
+    if (state == HTTP_PARSER_OPTIONAL_SP) {
+      ok = httpParserOnOptionalSp(parser, state, c, again);
+      break;
+    }
     if (state == HTTP_PARSER_STATUS_CODE) {
       ok = httpParserOnStatusCode(parser, state, c, again);
       break;

src/query/inc/qExtbuffer.h

@@ -77,13 +77,13 @@ typedef struct tFilePagesItem {
 
 typedef struct SSchemaEx {
   struct SSchema field;
-  int16_t        offset;
+  int32_t        offset;
 } SSchemaEx;
 
 typedef struct SColumnModel {
   int32_t    capacity;
   int32_t    numOfCols;
-  int16_t    rowSize;
+  int32_t    rowSize;
   SSchemaEx *pFields;
 } SColumnModel;
 

src/query/inc/qUtil.h

@@ -66,7 +66,8 @@ static FORCE_INLINE SResultRow *getResultRow(SResultRowInfo *pResultRowInfo, int
   return pResultRowInfo->pResult[slot];
 }
 
-static FORCE_INLINE char *getPosInResultPage(SQueryAttr *pQueryAttr, tFilePage* page, int32_t rowOffset, int16_t offset) {
+static FORCE_INLINE char* getPosInResultPage(SQueryAttr* pQueryAttr, tFilePage* page, int32_t rowOffset,
+                                             int32_t offset) {
   assert(rowOffset >= 0 && pQueryAttr != NULL);
 
   int32_t numOfRows = (int32_t)GET_ROW_PARAM_FOR_MULTIOUTPUT(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);

src/query/src/qAggMain.c

@@ -3982,7 +3982,7 @@ void blockInfo_func(SQLFunctionCtx* pCtx) {
 
   int32_t len = *(int32_t*) pCtx->pInput;
   blockDistInfoFromBinary((char*)pCtx->pInput + sizeof(int32_t), len, pDist);
-  pDist->rowSize = (int16_t) pCtx->param[0].i64;
+  pDist->rowSize = (uint16_t)pCtx->param[0].i64;
 
   memcpy(pCtx->pOutput, pCtx->pInput, sizeof(int32_t) + len);
 
@@ -4129,7 +4129,7 @@ void blockinfo_func_finalizer(SQLFunctionCtx* pCtx) {
   SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
   STableBlockDist* pDist = (STableBlockDist*) GET_ROWCELL_INTERBUF(pResInfo);
 
-  pDist->rowSize = (int16_t)pCtx->param[0].i64;
+  pDist->rowSize = (uint16_t)pCtx->param[0].i64;
   generateBlockDistResult(pDist, pCtx->pOutput);
 
   // cannot set the numOfIteratedElems again since it is set during previous iteration

src/query/src/qExecutor.c

@@ -3583,7 +3583,7 @@ void setResultRowOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pRe
   // Note: pResult->pos[i]->num == 0, there is only fixed number of results for each group
   tFilePage* bufPage = getResBufPage(pRuntimeEnv->pResultBuf, pResult->pageId);
 
-  int16_t offset = 0;
+  int32_t offset = 0;
   for (int32_t i = 0; i < numOfOutput; ++i) {
     pCtx[i].resultInfo = getResultCell(pResult, i, rowCellInfoOffset);
 
@@ -3897,7 +3897,7 @@ static int32_t doCopyToSDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SGroupResInfo*
 
     tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pRow->pageId);
 
-    int16_t offset = 0;
+    int32_t offset = 0;
     for (int32_t j = 0; j < pBlock->info.numOfCols; ++j) {
       SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, j);
       int32_t bytes = pColInfoData->info.bytes;

src/tsdb/inc/tsdbMemTable.h

@@ -71,27 +71,27 @@ int   tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxK
                             TKEY* filterKeys, int nFilterKeys, bool keepDup, SMergeInfo* pMergeInfo);
 void* tsdbCommitData(STsdbRepo* pRepo);
 
-static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
+static FORCE_INLINE SMemRow tsdbNextIterRow(SSkipListIterator* pIter) {
   if (pIter == NULL) return NULL;
 
   SSkipListNode* node = tSkipListIterGet(pIter);
   if (node == NULL) return NULL;
 
-  return (SDataRow)SL_GET_NODE_DATA(node);
+  return (SMemRow)SL_GET_NODE_DATA(node);
 }
 
 static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator* pIter) {
-  SDataRow row = tsdbNextIterRow(pIter);
+  SMemRow row = tsdbNextIterRow(pIter);
   if (row == NULL) return TSDB_DATA_TIMESTAMP_NULL;
 
-  return dataRowKey(row);
+  return memRowKey(row);
 }
 
 static FORCE_INLINE TKEY tsdbNextIterTKey(SSkipListIterator* pIter) {
-  SDataRow row = tsdbNextIterRow(pIter);
+  SMemRow row = tsdbNextIterRow(pIter);
   if (row == NULL) return TKEY_NULL;
 
-  return dataRowTKey(row);
+  return memRowTKey(row);
 }
 
 #endif /* _TD_TSDB_MEMTABLE_H_ */
\ No newline at end of file

src/tsdb/inc/tsdbMeta.h

@@ -32,7 +32,7 @@ typedef struct STable {
   void*          eventHandler;   // TODO
   void*          streamHandler;  // TODO
   TSKEY          lastKey;
-  SDataRow       lastRow;
+  SMemRow        lastRow;
   char*          sql;
   void*          cqhandle;
   SRWLatch       latch;  // TODO: implementa latch functions
@@ -148,7 +148,7 @@ static FORCE_INLINE STSchema *tsdbGetTableTagSchema(STable *pTable) {
 }
 
 static FORCE_INLINE TSKEY tsdbGetTableLastKeyImpl(STable* pTable) {
-  ASSERT(pTable->lastRow == NULL || pTable->lastKey == dataRowKey(pTable->lastRow));
+  ASSERT((pTable->lastRow == NULL) || (pTable->lastKey == memRowKey(pTable->lastRow)));
   return pTable->lastKey;
 }
 

src/tsdb/src/tsdbCommit.c

@@ -920,7 +920,8 @@ int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCo
     SDataCol * pDataCol = pDataCols->cols + ncol;
     SBlockCol *pBlockCol = pBlockData->cols + nColsNotAllNull;
 
-    if (isNEleNull(pDataCol, rowsToWrite)) {  // all data to commit are NULL, just ignore it
+    // if (isNEleNull(pDataCol, rowsToWrite)) {  // all data to commit are NULL, just ignore it
+    if (isAllRowsNull(pDataCol)) {  // all data to commit are NULL, just ignore it
       continue;
     }
 
@@ -1264,12 +1265,12 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
   while (true) {
     key1 = (*iter >= pDataCols->numOfRows) ? INT64_MAX : dataColsKeyAt(pDataCols, *iter);
     bool isRowDel = false;
-    SDataRow row = tsdbNextIterRow(pCommitIter->pIter);
-    if (row == NULL || dataRowKey(row) > maxKey) {
+    SMemRow row = tsdbNextIterRow(pCommitIter->pIter);
+    if (row == NULL || memRowKey(row) > maxKey) {
       key2 = INT64_MAX;
     } else {
-      key2 = dataRowKey(row);
-      isRowDel = dataRowDeleted(row);
+      key2 = memRowKey(row);
+      isRowDel = memRowDeleted(row);
     }
 
     if (key1 == INT64_MAX && key2 == INT64_MAX) break;
@@ -1284,24 +1285,24 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
       (*iter)++;
     } else if (key1 > key2) {
       if (!isRowDel) {
-        if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
-          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, dataRowVersion(row));
+        if (pSchema == NULL || schemaVersion(pSchema) != memRowVersion(row)) {
+          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row));
           ASSERT(pSchema != NULL);
         }
 
-        tdAppendDataRowToDataCol(row, pSchema, pTarget);
+        tdAppendMemRowToDataCol(row, pSchema, pTarget);
       }
 
       tSkipListIterNext(pCommitIter->pIter);
     } else {
       if (update) {
         if (!isRowDel) {
-          if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
-            pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, dataRowVersion(row));
+          if (pSchema == NULL || schemaVersion(pSchema) != memRowVersion(row)) {
+            pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row));
             ASSERT(pSchema != NULL);
           }
 
-          tdAppendDataRowToDataCol(row, pSchema, pTarget);
+          tdAppendMemRowToDataCol(row, pSchema, pTarget);
         }
       } else {
         ASSERT(!isRowDel);

src/tsdb/src/tsdbMain.c

@@ -641,7 +641,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
   int numColumns;
   int32_t blockIdx;
   SDataStatis* pBlockStatis = NULL;
-  SDataRow row = NULL;
+  SMemRow      row = NULL;
   // restore last column data with last schema
   
   int err = 0;
@@ -657,13 +657,15 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
     }
   }
 
-  row = taosTMalloc(dataRowMaxBytesFromSchema(pSchema));
+  row = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
   if (row == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     err = -1;
     goto out;
   }
-  tdInitDataRow(row, pSchema);
+
+  memRowSetType(row, SMEM_ROW_DATA);
+  tdInitDataRow(memRowDataBody(row), pSchema);
 
   // first load block index info
   if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
@@ -720,9 +722,9 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
       // OK,let's load row from backward to get not-null column
       for (int32_t rowId = pBlock->numOfRows - 1; rowId >= 0; rowId--) {
         SDataCol *pDataCol = pReadh->pDCols[0]->cols + i;
-        tdAppendColVal(row, tdGetColDataOfRow(pDataCol, rowId), pCol->type, pCol->bytes, pCol->offset);
+        tdAppendColVal(memRowDataBody(row), tdGetColDataOfRow(pDataCol, rowId), pCol->type, pCol->offset);
         //SDataCol *pDataCol = readh.pDCols[0]->cols + j;
-        void* value = tdGetRowDataOfCol(row, (int8_t)pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
+        void *value = tdGetRowDataOfCol(memRowDataBody(row), (int8_t)pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
         if (isNull(value, pCol->type)) {
           continue;
         }
@@ -742,8 +744,8 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
         // save row ts(in column 0)
         pDataCol = pReadh->pDCols[0]->cols + 0;
         pCol = schemaColAt(pSchema, 0);
-        tdAppendColVal(row, tdGetColDataOfRow(pDataCol, rowId), pCol->type, pCol->bytes, pCol->offset);
-        pLastCol->ts = dataRowKey(row);
+        tdAppendColVal(memRowDataBody(row), tdGetColDataOfRow(pDataCol, rowId), pCol->type, pCol->offset);
+        pLastCol->ts = memRowKey(row);
 
         pTable->restoreColumnNum += 1;
 
@@ -779,18 +781,18 @@ static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh,
   // Get the data in row
   
   STSchema *pSchema = tsdbGetTableSchema(pTable);
-  pTable->lastRow = taosTMalloc(dataRowMaxBytesFromSchema(pSchema));
+  pTable->lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
   if (pTable->lastRow == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
   }
-
-  tdInitDataRow(pTable->lastRow, pSchema);
+  memRowSetType(pTable->lastRow, SMEM_ROW_DATA);
+  tdInitDataRow(memRowDataBody(pTable->lastRow), pSchema);
   for (int icol = 0; icol < schemaNCols(pSchema); icol++) {
     STColumn *pCol = schemaColAt(pSchema, icol);
     SDataCol *pDataCol = pReadh->pDCols[0]->cols + icol;
-    tdAppendColVal(pTable->lastRow, tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type, pCol->bytes,
-                    pCol->offset);
+    tdAppendColVal(memRowDataBody(pTable->lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
+                   pCol->offset);
   }
 
   return 0;

src/tsdb/src/tsdbMemTable.c

@@ -21,7 +21,7 @@
 typedef struct {
   int32_t  totalLen;
   int32_t  len;
-  SDataRow row;
+  SMemRow  row;
 } SSubmitBlkIter;
 
 typedef struct {
@@ -36,20 +36,19 @@ static STableData *tsdbNewTableData(STsdbCfg *pCfg, STable *pTable);
 static void        tsdbFreeTableData(STableData *pTableData);
 static char *      tsdbGetTsTupleKey(const void *data);
 static int          tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables);
-static int          tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SDataRow row);
+static int              tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SMemRow row);
 static int          tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter);
-static SDataRow     tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
+static SMemRow          tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
 static int          tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg);
 static int          tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *affectedrows);
-static int          tsdbCopyRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable, void **ppRow);
+static int              tsdbCopyRowToMem(STsdbRepo *pRepo, SMemRow row, STable *pTable, void **ppRow);
 static int          tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter);
 static int          tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock);
 static int          tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable);
 static int          tsdbInsertDataToTableImpl(STsdbRepo *pRepo, STable *pTable, void **rows, int rowCounter);
 static void         tsdbFreeRows(STsdbRepo *pRepo, void **rows, int rowCounter);
-static int          tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SDataRow row);
-
-static FORCE_INLINE int tsdbCheckRowRange(STsdbRepo *pRepo, STable *pTable, SDataRow row, TSKEY minKey, TSKEY maxKey,
+static int              tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SMemRow row);
+static FORCE_INLINE int tsdbCheckRowRange(STsdbRepo *pRepo, STable *pTable, SMemRow row, TSKEY minKey, TSKEY maxKey,
                                           TSKEY now);
 
 int32_t tsdbInsertData(STsdbRepo *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pRsp) {
@@ -354,7 +353,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
   TSKEY      fKey = 0;
   bool       isRowDel = false;
   int        filterIter = 0;
-  SDataRow   row = NULL;
+  SMemRow    row = NULL;
   SMergeInfo mInfo;
 
   if (pMergeInfo == NULL) pMergeInfo = &mInfo;
@@ -365,12 +364,12 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
   if (pCols) tdResetDataCols(pCols);
 
   row = tsdbNextIterRow(pIter);
-  if (row == NULL || dataRowKey(row) > maxKey) {
+  if (row == NULL || memRowKey(row) > maxKey) {
     rowKey = INT64_MAX;
     isRowDel = false;
   } else {
-    rowKey = dataRowKey(row);
-    isRowDel = dataRowDeleted(row);
+    rowKey = memRowKey(row);
+    isRowDel = memRowDeleted(row);
   }
 
   if (filterIter >= nFilterKeys) {
@@ -407,12 +406,12 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
 
       tSkipListIterNext(pIter);
       row = tsdbNextIterRow(pIter);
-      if (row == NULL || dataRowKey(row) > maxKey) {
+      if (row == NULL || memRowKey(row) > maxKey) {
         rowKey = INT64_MAX;
         isRowDel = false;
       } else {
-        rowKey = dataRowKey(row);
-        isRowDel = dataRowDeleted(row);
+        rowKey = memRowKey(row);
+        isRowDel = memRowDeleted(row);
       }
     } else {
       if (isRowDel) {
@@ -437,12 +436,12 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
 
       tSkipListIterNext(pIter);
       row = tsdbNextIterRow(pIter);
-      if (row == NULL || dataRowKey(row) > maxKey) {
+      if (row == NULL || memRowKey(row) > maxKey) {
         rowKey = INT64_MAX;
         isRowDel = false;
       } else {
-        rowKey = dataRowKey(row);
-        isRowDel = dataRowDeleted(row);
+        rowKey = memRowKey(row);
+        isRowDel = memRowDeleted(row);
       }
 
       filterIter++;
@@ -548,7 +547,7 @@ static void tsdbFreeTableData(STableData *pTableData) {
   }
 }
 
-static char *tsdbGetTsTupleKey(const void *data) { return dataRowTuple((SDataRow)data); }
+static char *tsdbGetTsTupleKey(const void *data) { return memRowTuple((SMemRow)data); }
 
 static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables) {
   ASSERT(pMemTable->maxTables < maxTables);
@@ -572,17 +571,17 @@ static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables) {
   return 0;
 }
 
-static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SDataRow row) {
+static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SMemRow row) {
   if (pCols) {
-    if (*ppSchema == NULL || schemaVersion(*ppSchema) != dataRowVersion(row)) {
-      *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
+    if (*ppSchema == NULL || schemaVersion(*ppSchema) != memRowVersion(row)) {
+      *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row));
       if (*ppSchema == NULL) {
         ASSERT(false);
         return -1;
       }
     }
 
-    tdAppendDataRowToDataCol(row, *ppSchema, pCols);
+    tdAppendMemRowToDataCol(row, *ppSchema, pCols);
   }
 
   return 0;
@@ -592,31 +591,32 @@ static int tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
   if (pBlock->dataLen <= 0) return -1;
   pIter->totalLen = pBlock->dataLen;
   pIter->len = 0;
-  pIter->row = (SDataRow)(pBlock->data+pBlock->schemaLen);
+  pIter->row = (SMemRow)(pBlock->data + pBlock->schemaLen);
   return 0;
 }
 
-static SDataRow tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter) {
-  SDataRow row = pIter->row;
+static SMemRow tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter) {
+  SMemRow row = pIter->row;  // firstly, get current row
   if (row == NULL) return NULL;
 
-  pIter->len += dataRowLen(row);
-  if (pIter->len >= pIter->totalLen) {
+  pIter->len += memRowTLen(row);
+  if (pIter->len >= pIter->totalLen) {  // reach the end
     pIter->row = NULL;
   } else {
-    pIter->row = (char *)row + dataRowLen(row);
+    pIter->row = (char *)row + memRowTLen(row);  // secondly, move to next row
   }
 
   return row;
 }
 
-static FORCE_INLINE int tsdbCheckRowRange(STsdbRepo *pRepo, STable *pTable, SDataRow row, TSKEY minKey, TSKEY maxKey,
+static FORCE_INLINE int tsdbCheckRowRange(STsdbRepo *pRepo, STable *pTable, SMemRow row, TSKEY minKey, TSKEY maxKey,
                                           TSKEY now) {
-  if (dataRowKey(row) < minKey || dataRowKey(row) > maxKey) {
+  TSKEY rowKey = memRowKey(row);
+  if (rowKey < minKey || rowKey > maxKey) {
     tsdbError("vgId:%d table %s tid %d uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
               " maxKey %" PRId64 " row key %" PRId64,
               REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), now, minKey, maxKey,
-              dataRowKey(row));
+              rowKey);
     terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
     return -1;
   }
@@ -630,7 +630,7 @@ static int tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg) {
   SSubmitMsgIter msgIter = {0};
   SSubmitBlk *   pBlock = NULL;
   SSubmitBlkIter blkIter = {0};
-  SDataRow       row = NULL;
+  SMemRow        row = NULL;
   TSKEY          now = taosGetTimestamp(pRepo->config.precision);
   TSKEY          minKey = now - tsTickPerDay[pRepo->config.precision] * pRepo->config.keep;
   TSKEY          maxKey = now + tsTickPerDay[pRepo->config.precision] * pRepo->config.daysPerFile;
@@ -698,7 +698,7 @@ static int tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *
   int64_t        points = 0;
   STable *       pTable = NULL;
   SSubmitBlkIter blkIter = {0};
-  SDataRow       row = NULL;
+  SMemRow        row = NULL;
   void *         rows[TSDB_MAX_INSERT_BATCH] = {0};
   int            rowCounter = 0;
 
@@ -744,10 +744,10 @@ _err:
   return -1;
 }
 
-static int tsdbCopyRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable, void **ppRow) {
+static int tsdbCopyRowToMem(STsdbRepo *pRepo, SMemRow row, STable *pTable, void **ppRow) {
   STsdbCfg *  pCfg = &pRepo->config;
-  TKEY        tkey = dataRowTKey(row);
-  TSKEY       key = dataRowKey(row);
+  TKEY        tkey = memRowTKey(row);
+  TSKEY       key = memRowKey(row);
   bool        isRowDelete = TKEY_IS_DELETED(tkey);
 
   if (isRowDelete) {
@@ -765,15 +765,15 @@ static int tsdbCopyRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable, void
     }
   }
 
-  void *pRow = tsdbAllocBytes(pRepo, dataRowLen(row));
+  void *pRow = tsdbAllocBytes(pRepo, memRowTLen(row));
   if (pRow == NULL) {
-    tsdbError("vgId:%d failed to insert row with key %" PRId64 " to table %s while allocate %d bytes since %s",
-              REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), dataRowLen(row), tstrerror(terrno));
+    tsdbError("vgId:%d failed to insert row with key %" PRId64 " to table %s while allocate %" PRIu32 " bytes since %s",
+              REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), memRowTLen(row), tstrerror(terrno));
     return -1;
   }
 
-  dataRowCpy(pRow, row);
-  ppRow[0] = pRow;
+  memRowCpy(pRow, row);
+  ppRow[0] = pRow;  // save the memory address of data rows
 
   tsdbTrace("vgId:%d a row is %s table %s tid %d uid %" PRIu64 " key %" PRIu64, REPO_ID(pRepo),
             isRowDelete ? "deleted from" : "updated in", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable),
@@ -932,13 +932,15 @@ static int tsdbInsertDataToTableImpl(STsdbRepo *pRepo, STable *pTable, void **ro
   int64_t osize = SL_SIZE(pTableData->pData);
   tSkipListPutBatch(pTableData->pData, rows, rowCounter);
   int64_t dsize = SL_SIZE(pTableData->pData) - osize;
+  TSKEY   keyFirstRow = memRowKey(rows[0]);
+  TSKEY   keyLastRow = memRowKey(rows[rowCounter - 1]);
 
-  if (pMemTable->keyFirst > dataRowKey(rows[0])) pMemTable->keyFirst = dataRowKey(rows[0]);
-  if (pMemTable->keyLast < dataRowKey(rows[rowCounter - 1])) pMemTable->keyLast = dataRowKey(rows[rowCounter - 1]);
+  if (pMemTable->keyFirst > keyFirstRow) pMemTable->keyFirst = keyFirstRow;
+  if (pMemTable->keyLast < keyLastRow) pMemTable->keyLast = keyLastRow;
   pMemTable->numOfRows += dsize;
 
-  if (pTableData->keyFirst > dataRowKey(rows[0])) pTableData->keyFirst = dataRowKey(rows[0]);
-  if (pTableData->keyLast < dataRowKey(rows[rowCounter - 1])) pTableData->keyLast = dataRowKey(rows[rowCounter - 1]);
+  if (pTableData->keyFirst > keyFirstRow) pTableData->keyFirst = keyFirstRow;
+  if (pTableData->keyLast < keyLastRow) pTableData->keyLast = keyLastRow;
   pTableData->numOfRows += dsize;
 
   // update table latest info
@@ -954,8 +956,8 @@ static void tsdbFreeRows(STsdbRepo *pRepo, void **rows, int rowCounter) {
   STsdbBufPool *pBufPool = pRepo->pPool;
 
   for (int i = rowCounter - 1; i >= 0; --i) {
-    SDataRow row = (SDataRow)rows[i];
-    int      bytes = (int)dataRowLen(row);
+    SMemRow row = (SMemRow)rows[i];
+    int     bytes = (int)memRowTLen(row);
 
     if (pRepo->mem->extraBuffList == NULL) {
       STsdbBufBlock *pBufBlock = tsdbGetCurrBufBlock(pRepo);
@@ -988,21 +990,23 @@ static void tsdbFreeRows(STsdbRepo *pRepo, void **rows, int rowCounter) {
   }
 }
 
-static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SDataRow row) {
-  tsdbDebug("vgId:%d updateTableLatestColumn, %s row version:%d", REPO_ID(pRepo), pTable->name->data, dataRowVersion(row));
+static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow row) {
+  tsdbDebug("vgId:%d updateTableLatestColumn, %s row version:%d", REPO_ID(pRepo), pTable->name->data,
+            memRowVersion(row));
 
   STSchema* pSchema = tsdbGetTableLatestSchema(pTable);
   if (tsdbUpdateLastColSchema(pTable, pSchema) < 0) {
     return;
   }
 
-  pSchema = tsdbGetTableSchemaByVersion(pTable, dataRowVersion(row));
+  pSchema = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row));
   if (pSchema == NULL) {
     return;
   }
 
   SDataCol *pLatestCols = pTable->lastCols;
 
+  bool isDataRow = isDataRow(row);
   for (int16_t j = 0; j < schemaNCols(pSchema); j++) {
     STColumn *pTCol = schemaColAt(pSchema, j);
     // ignore not exist colId
@@ -1010,9 +1014,21 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SDataRow r
     if (idx == -1) {
       continue;
     }
-    
-    void* value = tdGetRowDataOfCol(row, (int8_t)pTCol->type, TD_DATA_ROW_HEAD_SIZE + pSchema->columns[j].offset);
-    if (isNull(value, pTCol->type)) {
+
+    void *value = NULL;
+
+    if (isDataRow) {
+      value = tdGetRowDataOfCol(memRowDataBody(row), (int8_t)pTCol->type,
+                                TD_DATA_ROW_HEAD_SIZE + pSchema->columns[j].offset);
+    } else {
+      // SKVRow
+      SColIdx *pColIdx = tdGetKVRowIdxOfCol(memRowKvBody(row), pTCol->colId);
+      if (pColIdx) {
+        value = tdGetKvRowDataOfCol(memRowKvBody(row), pColIdx->offset);
+      }
+    }
+
+    if ((value == NULL) || isNull(value, pTCol->type)) {
       continue;
     }
 
@@ -1027,11 +1043,11 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SDataRow r
 
     memcpy(pDataCol->pData, value, pDataCol->bytes);
     //tsdbInfo("updateTableLatestColumn vgId:%d cache column %d for %d,%s", REPO_ID(pRepo), j, pDataCol->bytes, (char*)pDataCol->pData);
-    pDataCol->ts = dataRowKey(row);
+    pDataCol->ts = memRowKey(row);
   }
 }
 
-static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SDataRow row) {
+static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SMemRow row) {
   STsdbCfg *pCfg = &pRepo->config;
 
   // if cacheLastRow config has been reset, free the lastRow
@@ -1042,31 +1058,31 @@ static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SDataRow
     TSDB_WUNLOCK_TABLE(pTable);
   }
 
-  if (tsdbGetTableLastKeyImpl(pTable) < dataRowKey(row)) {
+  if (tsdbGetTableLastKeyImpl(pTable) < memRowKey(row)) {
     if (CACHE_LAST_ROW(pCfg) || pTable->lastRow != NULL) {
-      SDataRow nrow = pTable->lastRow;
-      if (taosTSizeof(nrow) < dataRowLen(row)) {
-        SDataRow orow = nrow;
-        nrow = taosTMalloc(dataRowLen(row));
+      SMemRow nrow = pTable->lastRow;
+      if (taosTSizeof(nrow) < memRowTLen(row)) {
+        SMemRow orow = nrow;
+        nrow = taosTMalloc(memRowTLen(row));
         if (nrow == NULL) {
           terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
           return -1;
         }
 
-        dataRowCpy(nrow, row);
+        memRowCpy(nrow, row);
         TSDB_WLOCK_TABLE(pTable);
-        pTable->lastKey = dataRowKey(row);
+        pTable->lastKey = memRowKey(row);
         pTable->lastRow = nrow;
         TSDB_WUNLOCK_TABLE(pTable);
         taosTZfree(orow);
       } else {
         TSDB_WLOCK_TABLE(pTable);
-        pTable->lastKey = dataRowKey(row);
-        dataRowCpy(nrow, row);
+        pTable->lastKey = memRowKey(row);
+        memRowCpy(nrow, row);
         TSDB_WUNLOCK_TABLE(pTable);
       }
     } else {
-      pTable->lastKey = dataRowKey(row);
+      pTable->lastKey = memRowKey(row);
     }
 
     if (CACHE_LAST_NULL_COLUMN(pCfg)) {

src/tsdb/src/tsdbMeta.c

@@ -787,7 +787,7 @@ static char *getTagIndexKey(const void *pData) {
   void *    res = tdGetKVRowValOfCol(pTable->tagVal, pCol->colId);
   if (res == NULL) {
     // treat the column as NULL if we cannot find it
-    res = getNullValue(pCol->type);
+    res = (char*)getNullValue(pCol->type);
   }
   return res;
 }

src/tsdb/tests/tsdbTests.cpp

@@ -55,10 +55,10 @@ static int insertData(SInsertInfo *pInfo) {
       for (int j = 0; j < schemaNCols(pInfo->pSchema); j++) {
         STColumn *pTCol = schemaColAt(pInfo->pSchema, j);
         if (j == 0) {  // Just for timestamp
-          tdAppendColVal(row, (void *)(&start_time), pTCol->type, pTCol->bytes, pTCol->offset);
+          tdAppendColVal(row, (void *)(&start_time), pTCol->type, pTCol->offset);
         } else {  // For int
           int val = 10;
-          tdAppendColVal(row, (void *)(&val), pTCol->type, pTCol->bytes, pTCol->offset);
+          tdAppendColVal(row, (void *)(&val), pTCol->type, pTCol->offset);
         }
       }
       pBlock->dataLen += dataRowLen(row);

src/util/CMakeLists.txt

@@ -5,6 +5,7 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/rpc/inc)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/sync/inc)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/rmonotonic/inc)
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/TSZ/sz/include)
+
 AUX_SOURCE_DIRECTORY(src SRC)
 ADD_LIBRARY(tutil ${SRC})
 
@@ -14,7 +15,7 @@ TARGET_LINK_LIBRARIES(tutil pthread os lz4 z rmonotonic ${VAR_TSZ} )
 
 IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(tutil m rt)
-  # ADD_SUBDIRECTORY(tests)
+  ADD_SUBDIRECTORY(tests)
 
   FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/)
   IF (ICONV_INCLUDE_EXIST)
@@ -40,4 +41,4 @@ ENDIF()
 
 IF (TD_STORAGE)
   TARGET_LINK_LIBRARIES(tutil storage)
-ENDIF ()
+ENDIF ()
\ No newline at end of file

src/util/inc/tarray.h

@@ -52,6 +52,22 @@ void* taosArrayInit(size_t size, size_t elemSize);
  */
 void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
 
+/**
+ *
+ * @param pArray
+ * @param pData           position array list
+ * @param numOfElems      the number of removed position
+ */
+void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfElems);
+
+/**
+ *
+ * @param pArray
+ * @param comparFn
+ * @param fp
+ */
+void taosArrayRemoveDuplicate(SArray *pArray, __compar_fn_t comparFn, void (*fp)(void*));
+
 /**
  *  add all element from the source array list into the destination
  * @param pArray

src/util/src/tarray.c

@@ -83,6 +83,87 @@ void* taosArrayAddBatch(SArray* pArray, const void* pData, int nEles) {
   return dst;
 }
 
+void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfElems) {
+  assert(pArray != NULL && pData != NULL);
+  if (numOfElems <= 0) {
+    return;
+  }
+
+  size_t size = taosArrayGetSize(pArray);
+  if (numOfElems >= size) {
+    taosArrayClear(pArray);
+    return;
+  }
+
+  int32_t i = pData[0] + 1, j = 0;
+  while(i < size) {
+    if (j == numOfElems - 1) {
+      break;
+    }
+
+    char* p = TARRAY_GET_ELEM(pArray, i);
+    if (i > pData[j] && i < pData[j + 1]) {
+      char* dst = TARRAY_GET_ELEM(pArray, i - (j + 1));
+      memmove(dst, p, pArray->elemSize);
+    } else if (i == pData[j + 1]) {
+      j += 1;
+    }
+
+    i += 1;
+  }
+
+  assert(i == pData[numOfElems - 1] + 1);
+
+  int32_t dstIndex = pData[numOfElems - 1] - numOfElems + 1;
+  int32_t srcIndex = pData[numOfElems - 1] + 1;
+
+  char* dst = TARRAY_GET_ELEM(pArray, dstIndex);
+  char* src = TARRAY_GET_ELEM(pArray, srcIndex);
+  memmove(dst, src, pArray->elemSize * (pArray->size - numOfElems));
+
+  pArray->size -= numOfElems;
+}
+
+void taosArrayRemoveDuplicate(SArray *pArray, __compar_fn_t comparFn, void (*fp)(void*)) {
+  assert(pArray);
+
+  size_t size = pArray->size;
+  if (size <= 1) {
+    return;
+  }
+
+  int32_t pos = 0;
+  for(int32_t i = 1; i < size; ++i) {
+    char* p1 = taosArrayGet(pArray, pos);
+    char* p2 = taosArrayGet(pArray, i);
+
+    if (comparFn(p1, p2) == 0) {
+      // do nothing
+    } else {
+      if (pos + 1 != i) {
+        void* p = taosArrayGet(pArray, pos + 1);
+        if (fp != NULL) {
+          fp(p);
+        }
+
+        taosArraySet(pArray, pos + 1, p2);
+        pos += 1;
+      } else {
+        pos += 1;
+      }
+    }
+  }
+
+  if (fp != NULL) {
+    for(int32_t i = pos + 1; i < pArray->size; ++i) {
+      void* p = taosArrayGet(pArray, i);
+      fp(p);
+    }
+  }
+
+  pArray->size = pos + 1;
+}
+
 void* taosArrayAddAll(SArray* pArray, const SArray* pInput) {
   return taosArrayAddBatch(pArray, pInput->pData, (int32_t) taosArrayGetSize(pInput));
 }

src/util/src/terror.c

@@ -183,6 +183,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_FIELD_ALREAY_EXIST,       "Field already exists"
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_FIELD_NOT_EXIST,          "Field does not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STABLE_NAME,      "Super table does not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CREATE_TABLE_MSG, "Invalid create table message")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_EXCEED_MAX_ROW_BYTES,     "Exceed max row bytes")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_FUNC_NAME,        "Invalid func name")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_FUNC_LEN,         "Invalid func length")

src/util/tests/cacheTest.cpp

@@ -5,10 +5,6 @@
 #include "taos.h"
 #include "tcache.h"
 
-namespace {
-int32_t tsMaxMgmtConnections = 10000;
-int32_t tsMaxMeterConnections = 200;
-}
 // test cache
 TEST(testCase, client_cache_test) {
   const int32_t REFRESH_TIME_IN_SEC = 2;
@@ -43,7 +39,7 @@ TEST(testCase, client_cache_test) {
 
   sleep(3);
   char* d = (char*) taosCacheAcquireByKey(tscMetaCache, key3, strlen(key3));
-//    assert(d == NULL);
+  assert(d == NULL);
 
   char key5[] = "test5";
   char data5[] = "data5kkkkk";
@@ -102,7 +98,7 @@ TEST(testCase, cache_resize_test) {
 
   char key[256] = {0};
   char data[1024] = "abcdefghijk";
-  int32_t len = strlen(data);
+//  int32_t len = strlen(data);
 
   uint64_t startTime = taosGetTimestampUs();
   int32_t num = 10000;

tests/pytest/table/max_table_length.py

@@ -42,7 +42,7 @@ class TDTestCase:
         
         print("==============step3")
         tdLog.info("check int & binary")
-        tdSql.error("create table anal2 (ts timestamp ,i binary(16371),j int)")
+        # tdSql.error("create table anal2 (ts timestamp ,i binary(16371),j int)")
         tdSql.execute("create table anal2 (ts timestamp ,i binary(16370),j int)")
         tdSql.execute("create table anal3 (ts timestamp ,i binary(16366), j int, k int)")